View a markdown version of this page

停止執行中的工作階段 - Amazon Bedrock AgentCore
先決條件回應格式錯誤處理最佳實務

停止執行中的工作階段

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

無效的參數

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)

建議

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

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

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

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

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