Strands와 AgentCore를 활용해 Amazon RDS for SQL Server용 에이전틱 AI 구축하기

입력 → [추론 (LLM) → 도구 선택 → 도구 실행] → 응답
           ↑___________________↓

from strands import Agent, tool

@tool
def get_server_version() -> str:
    """SQL Server 버전을 반환합니다. 연결성을 확인하고
    엔진 버전을 식별하는 데 사용하세요."""
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT @@VERSION")
    version = cursor.fetchone()[0]
    conn.close()
    return version

from strands import Agent, tool
import boto3, pymssql, json, os
from datetime import datetime, timedelta

def get_db_connection():
    client = boto3.client('secretsmanager', region_name=os.getenv('AWS_REGION', 'us-west-2'))
    secret = client.get_secret_value(SecretId=os.getenv('DB_SECRET_ID'))
    creds = json.loads(secret['SecretString'])
    return pymssql.connect(
        server=creds['host'], user=creds['username'],
        password=creds['password'], port=creds['port']
    )

@tool
def get_deadlock_graphs(hours: int = 24) -> dict:
    """system_health 확장 이벤트 세션에서 교착 상태 그래프를 읽습니다.
    각 교착 상태의 양쪽에 대한 프로세스 목록, SQL 문, 잠금
    세부 정보가 포함된 XML 교착 상태 그래프를 반환합니다. 애플리케이션이
    오류 1205를 보고하거나 주기적인 교착 상태 분석에 사용하세요."""
    conn = get_db_connection()
    cursor = conn.cursor(as_dict=True)
    cursor.execute("""
        SELECT
            CAST(event_data AS XML).value('(event/@timestamp)[1]', 'DATETIME2') AS event_time,
            CAST(event_data AS XML).value('(event/data[@name="xml_report"]/value)[1]',
                'NVARCHAR(MAX)') AS deadlock_graph
        FROM sys.fn_xe_file_target_read_file(
            'd:\\rdsdbdata\\log\\system_health*.xel', NULL, NULL, NULL)
        WHERE CAST(event_data AS XML).value('(event/@name)[1]', 'VARCHAR(100)')
              = 'xml_deadlock_report'
          AND CAST(event_data AS XML).value('(event/@timestamp)[1]', 'DATETIME2')
              > DATEADD(HOUR, -%s, GETUTCDATE())
        ORDER BY event_time DESC
    """, (hours,))
    results = cursor.fetchall()
    conn.close()
    return {"deadlock_count": len(results), "deadlock_graphs": results}

@tool
def get_blocking_chains() -> dict:
    """sys.dm_exec_requests와 sys.dm_exec_sql_text를 사용하여 현재 차단 체인 계층을
    탐색합니다. 헤드 블로커, 해당 SQL, 대기 유형, 지속 시간, 그리고 모든 하위
    차단된 세션을 반환합니다. 애플리케이션이 타임아웃이나 행이 발생할 때 사용하세요."""
    conn = get_db_connection()
    cursor = conn.cursor(as_dict=True)
    cursor.execute("""
        WITH BlockingChain AS (
            SELECT r.session_id, r.blocking_session_id, r.wait_type,
                   r.wait_time / 1000.0 AS wait_seconds, r.status, r.command,
                   t.text AS sql_text, DB_NAME(r.database_id) AS database_name,
                   0 AS chain_level
            FROM sys.dm_exec_requests r
            CROSS APPLY sys.dm_exec_sql_text(r.sql_handle) t
            WHERE r.blocking_session_id = 0
              AND r.session_id IN (
                  SELECT DISTINCT blocking_session_id FROM sys.dm_exec_requests
                  WHERE blocking_session_id != 0)
            UNION ALL
            SELECT r.session_id, r.blocking_session_id, r.wait_type,
                   r.wait_time / 1000.0, r.status, r.command, t.text,
                   DB_NAME(r.database_id), bc.chain_level + 1
            FROM sys.dm_exec_requests r
            CROSS APPLY sys.dm_exec_sql_text(r.sql_handle) t
            JOIN BlockingChain bc ON r.blocking_session_id = bc.session_id
        )
        SELECT session_id, blocking_session_id, wait_type, wait_seconds,
               status, command, sql_text, database_name, chain_level
        FROM BlockingChain ORDER BY chain_level, wait_seconds DESC
    """)
    results = cursor.fetchall()
    conn.close()
    head_blockers = [r for r in results if r['chain_level'] == 0]
    return {"total_blocked_sessions": len([r for r in results if r['chain_level'] > 0]),
            "head_blockers": len(head_blockers), "blocking_chains": results}

@tool
def get_session_details(session_id: int) -> dict:
    """특정 세션의 로그인 이름, 호스트, 프로그램, 현재 SQL을 가져옵니다.
    헤드 블로커를 식별한 후에 사용하세요."""
    conn = get_db_connection()
    cursor = conn.cursor(as_dict=True)
    cursor.execute("""
        SELECT s.session_id, s.login_name, s.host_name, s.program_name,
               s.status, s.transaction_isolation_level,
               s.last_request_start_time, s.last_request_end_time,
               t.text AS current_sql
        FROM sys.dm_exec_sessions s
        OUTER APPLY sys.dm_exec_sql_text(s.most_recent_sql_handle) t
        WHERE s.session_id = %s
    """, (session_id,))
    result = cursor.fetchone()
    conn.close()
    return {"session_id": session_id, "details": result}

@tool
def send_diagnostic_report(subject: str, report: str) -> dict:
    """SNS 이메일을 통해 DBA 팀에 진단 보고서를 전송합니다. 조사를 완료한 후
    발견 사항과 권장 사항을 전달하는 데 사용하세요."""
    sns = boto3.client('sns', region_name=os.getenv('AWS_REGION', 'us-west-2'))
    topics = sns.list_topics()['Topics']
    topic_arn = next((t['TopicArn'] for t in topics
                      if os.getenv('SNS_TOPIC_NAME') in t['TopicArn']), None)
    if topic_arn:
        sns.publish(TopicArn=topic_arn, Subject=subject[:100], Message=report)
        return {"status": "sent", "topic": topic_arn}
    return {"status": "no_topic_found"}

@tool
def get_blocked_process_reports(hours: int = 24) -> dict:
    """확장 이벤트 파일 대상에서 차단된 프로세스 보고서를 읽습니다.
    SQL 텍스트, 대기 시간, 잠금 리소스를 포함한 블로커와 차단된 세션
    세부 정보를 모두 반환합니다. blocked_process_report를 캡처하는 사용자 정의
    XE 세션이 필요합니다. 잠금을 보유한 사람을 식별해야 할 때 과거 차단
    분석에 사용하세요."""
    conn = get_db_connection()
    cursor = conn.cursor(as_dict=True)
    cursor.execute("""
        SELECT
            CAST(event_data AS XML).value('(event/@timestamp)[1]', 'DATETIME2') AS event_time,
            CAST(event_data AS XML).value('(event/data[@name="duration"]/value)[1]', 'BIGINT') AS duration_ms,
            CAST(event_data AS XML).value('(event/data[@name="blocked_process"]/value)[1]',
                'NVARCHAR(MAX)') AS blocked_process_report
        FROM sys.fn_xe_file_target_read_file(
            'd:\\rdsdbdata\\log\\blocked*.xel', NULL, NULL, NULL)
        WHERE CAST(event_data AS XML).value('(event/@timestamp)[1]', 'DATETIME2')
              > DATEADD(HOUR, -%s, GETUTCDATE())
        ORDER BY event_time DESC
    """, (hours,))
    results = cursor.fetchall()
    conn.close()
    return {"blocked_process_count": len(results), "reports": results}

# ---------------------------------------------------------------------------
# AgentCore 앱 및 모델 구성
# ---------------------------------------------------------------------------
from strands import Agent, tool
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
import boto3, pymssql, json, os
from datetime import datetime, timedelta

app = BedrockAgentCoreApp()

model = BedrockModel(
    model_id=os.getenv('BEDROCK_MODEL_ID', '<your-model-id>'),
    region_name=os.getenv('AWS_REGION', '<your-region>'),
    temperature=0.3
)
# ---------------------------------------------------------------------------
# 데이터베이스 연결 헬퍼 — Secrets Manager에서 자격 증명을 검색
# ---------------------------------------------------------------------------

def get_db_connection():
    client = boto3.client('secretsmanager', region_name=os.getenv('AWS_REGION', '<your-region>'))
    secret = client.get_secret_value(SecretId=os.getenv('DB_SECRET_ID'))
    creds = json.loads(secret['SecretString'])
    return pymssql.connect(
        server=creds['host'], user=creds['username'],
        password=creds['password'], port=creds['port']
    )

# --- 여기에 도구 코드 붙여넣기 ---
# @tool get_deadlock_graphs       (교착 상태 탐지 섹션에서)
# @tool get_blocking_chains       (차단 체인 분석 섹션에서)
# @tool get_session_details       (세션 세부 정보 섹션에서)
# @tool get_blocked_process_reports (과거 차단된 프로세스 섹션에서)
# @tool send_diagnostic_report    (진단 보고서 섹션에서)

# ---------------------------------------------------------------------------
# 에이전트 정의 — 시스템 프롬프트가 추론을 안내하고, 도구가 기능을 정의
# ---------------------------------------------------------------------------
agent = Agent(
    system_prompt="""당신은 Amazon RDS용 SQL Server DBOps 에이전트입니다.
문제를 조사할 때:
1. 증상부터 시작하세요 — 모든 질문에 모든 도구를 실행하지 마세요
2. 애플리케이션이 오류 1205나 트랜잭션 실패를 보고할 때 교착 상태 도구를 사용하세요
3. 애플리케이션이 타임아웃이나 잠금 대기를 보고할 때 차단 체인 도구를 사용하세요
4. 현재 차단이 해결된 상태에서 과거 차단 분석을 위해 차단된 프로세스 보고서를 사용하세요
5. 평가를 제공하기 전에 데이터 소스 간의 발견 사항을 연관시키세요
6. 심각도(Critical/Warning/Info)와 구체적이고 실행 가능한 권장 사항을 제공하세요
7. SNS를 통해 진단 보고서를 전송할 때 다음을 포함하세요: 영향받은 세션 ID, SQL 문, 대기 유형, 지속 시간, 호스트/IP, 로그인 이름, 근본 원인, 구체적인 수정 단계""",
    model=model,
    tools=[get_deadlock_graphs, get_blocking_chains, get_session_details,
           get_blocked_process_reports, send_diagnostic_report]
)
# ---------------------------------------------------------------------------
# AgentCore 진입점 — agentcore invoke로부터 페이로드를 수신
# ---------------------------------------------------------------------------

@app.entrypoint
def handler(payload):
    response = agent(payload.get("prompt", ""))
    return response.message['content'][0]['text']

if __name__ == "__main__":
    app.run()

git clone https://github.com/aws-samples/sample-agentcore-sqlserver-dbops-agent.git
cd sample-agentcore-sqlserver-dbops-agent

uv venv && source .venv/bin/activate
uv pip install -r requirements.txt
uv pip install bedrock-agentcore-starter-toolkit

export AWS_REGION=<your-region>
export DB_INSTANCE_ID=<your-rds-instance-id>       # RDS SQL Server 인스턴스
export DB_SECRET_ID=<your-secrets-id>            # Secrets Manager 자격 증명
export SNS_TOPIC_NAME=<your-sns-topic-name>         # 보고서용 SNS 토픽
export AGENTCORE_ROLE_ARN=<your-execution-role-arn>    # IAM Role(역할)
export SECURITY_GROUP_ID=<your-security-group-id>      # VPC 보안 그룹
export SUBNET1=<your-first-subnet-id>                  # VPC 프라이빗 서브넷
export SUBNET2=<your-second-subnet-id>                 # VPC 프라이빗 서브넷

agentcore configure \
  --name my_first_agentcore_runtime \
  --entrypoint agent.py \
  --execution-role $AGENTCORE_ROLE_ARN \
  --deployment-type direct_code_deploy \
  --vpc \
  --subnets $SUBNET1,$SUBNET2 \
  --security-groups $SECURITY_GROUP_ID

agentcore deploy \
  --env AWS_REGION=$AWS_REGION \
  --env DB_SECRET_ID=$DB_SECRET_ID \
  --env SNS_TOPIC_NAME=$SNS_TOPIC_NAME \
  --env AGENT_OBSERVABILITY_ENABLED=true

agentcore status

agentcore invoke '{"prompt": "Check for deadlocks in the last 24 hours, provide RCA and recommendations, and send a diagnostic report via SNS"}'

demo-server $
demo-server $ agentcore invoke '{"prompt": "Check for deadlocks in the last 24 hours, provide RCA and recommendations, and send a diagnostic report via SNS"}'

╭───────────────────────────────────────────────────────────────────── my_first_agentcore_runtime ─────────────────────────────────────────────────────────────────────╮
│ Session: bb18d342-0987-4854-bc8f-bb5c0edff095                                                                                                                       │
│ Request ID: 554ff565-e99e-430c-b79c-a6e480d14a06                                                                                                                    │
│ ARN: arn:aws:bedrock-agentcore:us-west-2:111122223333:runtime/my_first_agentcore_runtime-MnC7NP39B6                                                                 │
│ Logs: aws logs tail /aws/bedrock-agentcore/runtimes/my_first_agentcore_runtime-MnC7NP39B6-DEFAULT --log-stream-name-prefix "2026/03/25/[runtime-logs" --follow       │
│       aws logs tail /aws/bedrock-agentcore/runtimes/my_first_agentcore_runtime-MnC7NP39B6-DEFAULT --log-stream-name-prefix "2026/03/25/[runtime-logs" --since 1h     │
│ GenAI Dashboard: https://console.aws.amazon.com/cloudwatch/home?region=us-west-2#gen-ai-observability/agent-core                                                    │
╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯

Response:

## 종합 분석 완료

철저한 교착 상태 조사를 수행하고 상세한 진단 보고서를 전달했습니다. 다음은 제 평가입니다:

### 주요 발견 사항:
- 2026년 3월 25일 07:16:01에 **1건의 교착 상태** 탐지
- **심각도: WARNING** - 빈도는 낮지만 사전 개입이 필요함
- **근본 원인:** Suppliers와 Employees 테이블 업데이트 간의 잠금 순서 불일치

### 근본 원인 분석:
교착 상태는 관련 레코드(둘 다 ID = 1)를 다른 순서로 업데이트하는 동시 트랜잭션으로 인해
발생하여 순환 대기 조건을 생성했습니다. 이는 다음 중 하나를 시사합니다:
- 테이블 간의 외래 키 관계
- 일관된 잠금 순서를 따르지 않는 애플리케이션 로직
- 공급업체와 직원 데이터 간의 잠재적인 비즈니스 프로세스 결합

### 전달된 중요 권장 사항:
1. **즉시:** 교착 상태 재시도 로직 및 일관된 잠금 순서 구현
2. **단기:** READ_COMMITTED_SNAPSHOT 활성화 및 모니터링 강화
3. **장기:** 아키텍처 검토 및 성능 최적화

### 완료된 작업:
✅ 종합적인 교착 상태 분석
✅ 상세한 근본 원인 분석
✅ 타임라인이 포함된 우선순위 권장 사항
✅ 예방 전략 수립
✅ 모니터링 계획 수립
✅ DBA 팀에 SNS를 통한 진단 보고서 전송

보고서에는 향후 발생을 방지하기 위한 구체적인 기술적 세부 사항, 비즈니스 영향 평가,
우선순위와 타임라인이 포함된 구조화된 실행 계획이 포함되어 있습니다.

demo-server $

agentcore memory create dbops_shared_memory \
  --strategies '[{"semanticMemoryStrategy": {"name": "dbops_facts"}}, {"summaryMemoryStrategy": {"name": "dbops_summaries"}}]' \
  --event-expiry-days 30 \
  --region $AWS_REGION \
  --wait
AGENTCORE_MEMORY_ID=$(agentcore memory list --region $AWS_REGION 2>/dev/null | grep dbops_shared_memory | awk '{print $4}')

agentcore destroy --agent my_first_agentcore_runtime

agentcore memory delete $AGENTCORE_MEMORY_ID --region $AWS_REGION