停止正在运行的会话

该StopRuntimeSession操作允许您立即终止活动的代理 AgentCore 运行时会话，以便进行适当的资源清理和会话生命周期管理。

调用时，此操作会立即终止指定的会话并停止所有正在进行的流媒体响应。这样可以正确释放系统资源并防止孤立会话的积累。

StopRuntimeSession在以下场景中使用：

• User-initiated 结束：当用户明确结束对话时

• 应用程序关闭：应用程序终止前的主动清理

• 错误处理：强制终止无响应或停滞的会话

• 配额管理：通过关闭未使用的会话保持在会话限制之内

• 超时处理：清理超过预期持续时间的会话

先决条件

要使用 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 SDK 会自动重试此异常。如果您禁用了重试或在没有 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)

建议

• 停止会话时务必处理异常

• 跟踪应用程序中的活动会话以进行清理

• 为停止请求设置超时以避免挂起

• 记录会话终止以进行调试和监控

• 使用会话管理器管理具有多个会话的复杂应用程序