View a markdown version of this page

Interrompere una sessione in esecuzione - Amazon Bedrock AgentCore

Interrompere una sessione in esecuzione

L'StopRuntimeSessionoperazione consente di terminare immediatamente le sessioni di Active Agent AgentCore Runtime per una corretta pulizia delle risorse e una corretta gestione del ciclo di vita della sessione.

Quando viene richiamata, questa operazione termina istantaneamente la sessione specificata e interrompe tutte le risposte di streaming in corso. Ciò consente alle risorse di sistema di essere rilasciate correttamente e impedisce l'accumulo di sessioni orfane.

Utilizza StopRuntimeSession in questi scenari:

  • User-initiated fine: quando gli utenti concludono esplicitamente la conversazione

  • Chiusura dell'applicazione: pulizia proattiva prima della chiusura dell'applicazione

  • Gestione degli errori: interruzione forzata delle sessioni che non rispondono o sono bloccate

  • Gestione delle quote: rimani entro i limiti delle sessioni chiudendo le sessioni non utilizzate

  • Gestione del timeout: ripulisci le sessioni che superano la durata prevista

Prerequisiti

Per utilizzarlo StopRuntimeSession, è necessario:

  • Autorizzazione IAM bedrock-agentcore:StopRuntimeSession

  • Un agente AgentCore Runtime ARN valido

  • L'ID di una sessione attiva da terminare

Esempio
AWS SDK
  1. Puoi utilizzare l' AWS SDK per interrompere le sessioni di AgentCore Runtime a livello di codice.

    Esempio in Python che utilizza boto3 per interrompere una sessione di Runtime. 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. Per le applicazioni che utilizzano l'autenticazione OAuth, effettua richieste HTTPS dirette:

    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 della risposta

Formato di risposta previsto per operazioni riusciteStopRuntimeSession.

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

Gestione degli errori

Risposte di errore comuni:

Codice di stato Errore Description

404

ResourceNotFoundException

Sessione non trovata o già terminata

403

AccessDeniedException

Autorizzazioni insufficienti

400

ValidationException

Parametri non validi

500

InternalServerException

Errore di servizio

Best practice

Gestione del ciclo di vita della sessione

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)

Raccomandazioni

  • Gestisci sempre le eccezioni quando interrompi le sessioni

  • Tieni traccia delle sessioni attive nell'applicazione per la pulizia

  • Imposta dei timeout per le richieste di interruzione per evitare interruzioni

  • Registra le terminazioni delle sessioni per il debug e il monitoraggio

  • Utilizza i gestori di sessione per applicazioni complesse con più sessioni