停止執行中的工作階段

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	無效的參數
409	RetryableConflictException	工作階段操作進行中，請重試
500	InternalServerException	服務錯誤

當您的StopRuntimeSession呼叫在服務仍在佈建或銷毀工作階段時到達時，服務會傳回 RetryableConflictException(HTTP 409)。此條件為暫時性且可重試。以短指數退避重試，而不是將其視為終端故障。啟用預設重試時 AWS SDKs 會自動重試此例外狀況。如果您停用重試或直接呼叫 API，而沒有 AWS SDK （例如，稍早顯示的 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)

建議

• 停止工作階段時一律處理例外狀況

• 追蹤應用程式中的作用中工作階段以進行清除

• 設定停止請求的逾時，以避免掛起

• 用於偵錯和監控的日誌工作階段終止

• 針對具有多個工作階段的複雜應用程式使用工作階段管理員