Amazon Bedrock 기반 자동 근본 원인 분석 파이프라인

import base64
import boto3
import requests
from strands import Agent, tool

# Retrieve the GitHub token from AWS Secrets Manager
secrets_client = boto3.client("secretsmanager")
github_token = secrets_client.get_secret_value(
    SecretId="trends/github-token"
)["SecretString"]

@tool
def fetch_source_code(file_path: str, repo: str) -> str:
    """Fetch a source file from a GitHub repository.
    Args: file_path: Path to the file in the repository repo: Repository in 'owner/repo' format
    """
    response = requests.get(
        f"https://api.github.com/repos/{repo}/contents/{file_path}",
        headers={"Authorization": f"token {github_token}"}
    )
    if response.status_code != 200:
        return f"Could not fetch {file_path} from {repo}: HTTP {response.status_code}"
    content = base64.b64decode(response.json()["content"])
    return content.decode("utf-8")

@tool
def fetch_log_context(log_group: str, log_stream: str, timestamp: int, window_seconds: int = 30) -> str:
    """Fetch log lines from the same log stream surrounding an error.
    Args: log_group: CloudWatch Log Group name
          log_stream: Log stream that produced the error
          timestamp: Error event timestamp in milliseconds
          window_seconds: Time window before and after the error
    """
    response = logs_client.filter_log_events(
        logGroupName=log_group,
        logStreamNames=[log_stream],
        startTime=timestamp - (window_seconds * 1000),
        endTime=timestamp + (window_seconds * 1000),
    )
    events = response.get("events", [])
    if not events:
        return f"No log events found in {log_stream} within {window_seconds}s of the error."
    return "
".join(e["message"] for e in events)