View a markdown version of this page

停止正在运行的会话 - Amazon Bedrock AgentCore
先决条件响应格式错误处理最佳实践

停止正在运行的会话

该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

参数无效

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)

建议

  • 停止会话时@@ 一定要处理异常

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

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

  • 记录会话终止情况，以便进行调试和监控

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