실행 중인 세션 중지

StopRuntimeSession 작업을 통해 적절한 리소스 정리 및 세션 수명 주기 관리를 위해 활성 에이전트 AgentCore 런타임 세션을 즉시 종료할 수 있습니다.

호출되면이 작업은 지정된 세션을 즉시 종료하고 진행 중인 스트리밍 응답을 중지합니다. 이렇게 하면 시스템 리소스가 제대로 릴리스되고 분리된 세션이 누적되는 것을 방지할 수 있습니다.

다음 시나리오StopRuntimeSession에서를 사용합니다.

• 사용자 시작 종료: 사용자가 대화를 명시적으로 종료하는 경우

• 애플리케이션 종료: 애플리케이션 종료 전 사전 예방적 정리

• 오류 처리: 응답하지 않거나 중지된 세션 강제 종료

• 할당량 관리: 사용하지 않는 세션을 닫아 세션 제한 내에서 유지

• 제한 시간 처리: 예상 기간을 초과하는 세션 정리

사전 조건

StopRuntimeSession을 사용하려면 다음이 필요합니다.

• bedrock-agentcore:StopRuntimeSession IAM 권한

• 유효한 에이전트 AgentCore 런타임 ARN

• 종료할 활성 세션의 ID입니다.

예

AWS SDK

1. AWS SDK를 사용하여 프로그래밍 방식으로 AgentCore 런타임 세션을 중지할 수 있습니다.

   boto3를 사용하여 AgentCore 런타임 세션을 중지하는 Python 예제입니다.

   import boto3
   
   # Initialize the AgentCore client
   client = boto3.client('bedrock-agentcore', region_name='us-west-2')
   
   try:
       # Stop the runtime session
       response = client.stop_runtime_session(
           agentRuntimeArn='arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-agent',
           runtimeSessionId='your-session-id',
           qualifier='DEFAULT'  # Optional: endpoint name
       )
   
       print(f"Session terminated successfully")
       print(f"Request ID: {response['ResponseMetadata']['RequestId']}")
   
   except client.exceptions.ResourceNotFoundException:
       print("Session not found or already terminated")
   except client.exceptions.AccessDeniedException:
       print("Insufficient permissions to stop session")
   except Exception as e:
       print(f"Error stopping session: {str(e)}")

HTTPS request

1. OAuth 인증을 사용하는 애플리케이션의 경우 직접 HTTPS 요청을 수행합니다.

   import requests
   import urllib.parse
   
   def stop_session_with_oauth(agent_arn, session_id, bearer_token, qualifier="DEFAULT"):
       # URL encode the agent ARN
       encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
   
       # Construct the endpoint URL
       url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/stopruntimesession?qualifier={qualifier}"
   
       headers = {
           "Authorization": f"Bearer {bearer_token}",
           "Content-Type": "application/json",
           "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": session_id
       }
   
       try:
           response = requests.post(url, headers=headers)
           response.raise_for_status()
   
           print(f"Session {session_id} terminated successfully")
           return response.json()
   
       except requests.exceptions.HTTPError as e:
           if e.response.status_code == 404:
               print("Session not found or already terminated")
           elif e.response.status_code == 403:
               print("Insufficient permissions or invalid token")
           else:
               print(f"HTTP error: {e.response.status_code}")
           raise
       except requests.exceptions.RequestException as e:
           print(f"Request failed: {str(e)}")
           raise
   
   # Usage
   stop_session_with_oauth(
       agent_arn="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-agent",
       session_id="your-session-id",
       bearer_token="your-oauth-token"
   )

응답 형식

성공적인 StopRuntimeSession 작업을 위한 예상 응답 형식입니다.

{
    "ResponseMetadata": {
        "RequestId": "12345678-1234-1234-1234-123456789012",
        "HTTPStatusCode": 200
    }
}

오류 처리

일반적인 오류 응답:

상태 코드	오류	설명
404	ResourceNotFoundException	세션을 찾을 수 없거나 이미 종료됨
403	AccessDeniedException	권한 부족
400	ValidationException	잘못된 파라미터
409	RetryableConflictException	세션 작업이 진행 중입니다. 다시 시도하세요.
500	InternalServerException	서비스 오류

서비스가 세션을 프로비저닝하거나 해제하는 동안 StopRuntimeSession 통화가 도착하면 서비스는 RetryableConflictException (HTTP 409)를 반환합니다. 이 조건은 일시적이며 재시도 가능합니다. 터미널 장애로 처리하는 대신 짧은 지수 백오프로 재시도합니다. 기본 재시도가 활성화된 경우 AWS SDKs 자동 재시도합니다. 재시도를 비활성화하거나 AWS SDK 없이 직접 API를 호출하는 경우(예: 앞에서 설명한 HTTPS 요청) 직접 재시도합니다.

모범 사례

세션 수명 주기 관리

class SessionManager:
    def __init__(self, client, agent_arn):
        self.client = client
        self.agent_arn = agent_arn
        self.active_sessions = set()

    def invoke_agent(self, session_id, payload):
        """Invoke agent and track session"""
        try:
            response = self.client.invoke_agent_runtime(
                agentRuntimeArn=self.agent_arn,
                runtimeSessionId=session_id,
                payload=payload
            )
            self.active_sessions.add(session_id)
            return response
        except Exception as e:
            print(f"Failed to invoke agent: {e}")
            raise

    def stop_session(self, session_id):
        """Stop session and remove from tracking"""
        try:
            self.client.stop_runtime_session(
                agentRuntimeArn=self.agent_arn,
                runtimeSessionId=session_id,
                qualifier=endpoint_name
            )
            self.active_sessions.discard(session_id)
            print(f"Session {session_id} stopped")
        except Exception as e:
            print(f"Failed to stop session {session_id}: {e}")

    def cleanup_all_sessions(self):
        """Stop all tracked sessions"""
        for session_id in list(self.active_sessions):
            self.stop_session(session_id)

권장 사항

• 세션을 중지할 때 항상 예외 처리

• 정리를 위해 애플리케이션의 활성 세션 추적

• 중단되지 않도록 중지 요청에 대한 제한 시간 설정

• 디버깅 및 모니터링을 위한 로그 세션 종료

• 여러 세션이 있는 복잡한 애플리케이션에 세션 관리자 사용