Interromper uma sessão em execução

A StopRuntimeSession operação permite que você encerre imediatamente as sessões ativas do agente AgentCore Runtime para a limpeza adequada dos recursos e o gerenciamento do ciclo de vida da sessão.

Quando chamada, essa operação encerra instantaneamente a sessão especificada e interrompe todas as respostas de streaming em andamento. Isso permite que os recursos do sistema sejam liberados adequadamente e evita o acúmulo de sessões órfãs.

Use StopRuntimeSession nesses cenários:

• User-initiated fim: quando os usuários concluem explicitamente a conversa

• Desligamento do aplicativo: limpeza proativa antes do encerramento do aplicativo

• Tratamento de erros: forçar o encerramento de sessões sem resposta ou paralisadas

• Gerenciamento de cotas: permaneça dentro dos limites da sessão fechando sessões não utilizadas

• Tratamento do tempo limite: limpe as sessões que excedem a duração esperada

Pré-requisitos

Para usar StopRuntimeSession, você precisa:

• Permissão bedrock-agentcore:StopRuntimeSession do IAM

• Um ARN de AgentCore tempo de execução de agente válido

• O ID de uma sessão ativa a ser encerrada

exemplo

AWS SDK

1. Você pode usar o AWS SDK para interromper as sessões do AgentCore Runtime de forma programática.

   Exemplo em Python usando boto3 para interromper uma sessão de tempo de execução. AgentCore

   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. Para aplicativos que usam a autenticação OAuth, faça solicitações HTTPS diretas:

   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"
   )

Formato de resposta

Formato de resposta esperado para StopRuntimeSession operações bem-sucedidas.

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

Tratamento de erros

Respostas de erro comuns:

Código de status	Erro	Description
404	ResourceNotFoundException	Sessão não encontrada ou já foi encerrada
403	AccessDeniedException	Permissões insuficientes
400	ValidationException	Parâmetros inválidos
500	InternalServerException	Erro de serviço

Práticas recomendadas

Gerenciamento do ciclo de vida da sessão

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)

Recomendações

• Sempre manipule as exceções ao interromper as sessões

• Acompanhe as sessões ativas em seu aplicativo para limpeza

• Defina tempos limite para solicitações de interrupção para evitar interrupções

• Registre as terminações da sessão para depuração e monitoramento

• Use gerenciadores de sessão para aplicativos complexos com várias sessões