View a markdown version of this page

範例 - Amazon Bedrock AgentCore

本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。

範例

下列範例顯示常見攔截器使用案例的 Python AWS Lambda 函數。

傳遞攔截器

此範例示範一個簡單的攔截器,記錄 REQUEST 攔截器的 MCP 方法,並透過不變的方式傳遞所有請求和回應:

import json import logging # Configure logging logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): """ Lambda function that handles both REQUEST and RESPONSE interceptor types. For REQUEST interceptors: logs the MCP method and passes request through unchanged For RESPONSE interceptors: passes response through unchanged """ # Extract the MCP data from the event mcp_data = event.get('mcp', {}) # Check if this is a REQUEST or RESPONSE interceptor based on presence of gatewayResponse if 'gatewayResponse' in mcp_data and mcp_data['gatewayResponse'] != None: # This is a RESPONSE interceptor logger.info("Processing RESPONSE interceptor - passing through unchanged") # Pass through the original request and response unchanged response = { "interceptorOutputVersion": "1.0", "mcp": { "transformedGatewayResponse": { "body": mcp_data.get('gatewayResponse', {}).get('body', {}), "statusCode": mcp_data.get('gatewayResponse', {}).get('statusCode', 200) } } } else: # This is a REQUEST interceptor gateway_request = mcp_data.get('gatewayRequest', {}) request_body = gateway_request.get('body', {}) mcp_method = request_body.get('method', 'unknown') # Log the MCP method logger.info(f"Processing REQUEST interceptor - MCP method: {mcp_method}") # Pass through the original request unchanged response = { "interceptorOutputVersion": "1.0", "mcp": { "transformedGatewayRequest": { "body": request_body, } } } return response

此 Lambda 函數可以同時設定為 REQUEST 和 RESPONSE 攔截器。當設定為 REQUEST 攔截器時,它會記錄來自傳入請求的 MCP 方法。設定為 RESPONSE 攔截器時,只會傳遞未變更的回應。這兩種攔截器類型都會在不修改的情況下傳回原始資料,使其成為「傳遞」攔截器。

使用請求攔截器自訂模型路由

對於推論目標,閘道會使用模型型路由從 model 欄位選取目標。REQUEST 攔截器可以在閘道評估路由model之前重寫,這可讓您支援虛擬模型 ID:未對應至任何單一設定模型的穩定別名。攔截器會將別名解析為格式為 的具體目標合格 ID{targetName}/{modelId},因此呼叫者會在您集中控制模型選擇時使用一個名稱。

HTTP 攔截器承載

推論目標使用http攔截器承載,其中請求內文是 base64 編碼字串。如需詳細資訊,請參閱 HTTP 目標的攔截器。

下列 REQUEST 攔截器會根據請求輸入的大小,將虛擬模型 ID 解析auto-claude為特定 Anthropic 模型:anthropic/claude-haiku-4-5適用於小型請求、anthropic/claude-sonnet-4-6適用於中型請求,以及anthropic/claude-opus-4-7適用於大型請求。呼叫者傳送 auto-claude,而閘道會將每個請求路由到anthropic目標上已解析的模型。使用任何其他模型 ID 的請求會傳遞不變,並遵循正常的模型型路由。

import base64 import json import logging logger = logging.getLogger() logger.setLevel(logging.INFO) # The virtual model ID that this interceptor resolves to a concrete # target-qualified ID in the form "{targetName}/{modelId}". VIRTUAL_MODEL = "auto-claude" # Approximate input-size thresholds (in characters) used to choose a model. SONNET_THRESHOLD = 2000 OPUS_THRESHOLD = 8000 def resolve_auto_claude(payload): """Choose an Anthropic model based on the size of the request input.""" input_size = len(json.dumps(payload.get("input", ""))) if input_size >= OPUS_THRESHOLD: return "anthropic/claude-opus-4-7" if input_size >= SONNET_THRESHOLD: return "anthropic/claude-sonnet-4-6" return "anthropic/claude-haiku-4-5" def lambda_handler(event, context): http = event.get("http", {}) request = http.get("gatewayRequest", {}) encoded_body = request.get("body") # Nothing to transform (for example, an empty body) - pass through. if not encoded_body: return {"interceptorOutputVersion": "1.0", "http": {}} try: payload = json.loads(base64.b64decode(encoded_body)) except (ValueError, TypeError) as exc: logger.warning("Passing request through unchanged; could not parse body: %s", exc) return {"interceptorOutputVersion": "1.0", "http": {}} # Only resolve the virtual model ID; let everything else route normally. if not isinstance(payload, dict) or payload.get("model") != VIRTUAL_MODEL: return {"interceptorOutputVersion": "1.0", "http": {}} resolved = resolve_auto_claude(payload) logger.info("Resolved virtual model %s -> %s", VIRTUAL_MODEL, resolved) payload["model"] = resolved encoded = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8") return { "interceptorOutputVersion": "1.0", "http": { "transformedGatewayRequest": { "body": encoded } }, }

將 函數連接至您的閘道做為 REQUEST 攔截器。如需說明,請參閱設定攔截器。閘道服務角色也必須具有叫用 函數的許可。如需詳細資訊,請參閱攔截器的許可。

若要查看解決方案,請傳送使用auto-claude別名的請求。攔截器會根據輸入大小重寫model至具體的 Anthropic 模型,而閘道會將請求路由至anthropic目標上的該模型。

awscurl --service bedrock-agentcore --region us-west-2 -X POST \ "https://GATEWAY_ID.gateway.bedrock-agentcore.us-west-2.amazonaws.com/inference/v1/responses" \ -H "Content-Type: application/json" \ -d '{"model": "auto-claude", "input": "Hello!", "max_output_tokens": 50}'