View a markdown version of this page

실행 중인 세션 중지 - Amazon Bedrock AgentCore
사전 조건응답 형식오류 처리모범 사례

실행 중인 세션 중지

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

잘못된 파라미터

500

InternalServerException

서비스 오류

모범 사례

세션 수명 주기 관리

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)

권장 사항

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

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

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

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

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