Amazon Bedrock AgentCore Gateway

AgentCore Gateway runs a Lambda interceptor around every MCP call it proxies to a tool server: once on the request (tool arguments) and once on the response (tool result). The interceptor below is the scaffold that receives both hooks, holds the Alice client, and returns the transformed body the gateway forwards. Prompts and model responses never pass through the gateway, so they are out of scope here.

Status: Preview · Evaluates: Tool calls · Vendor: AWS

MCP tools/call arguments and tool results at the gateway.

Note: This example uses the v1 WonderFenceClient interface of the WonderFence SDK.

Setup

Configuration: ALICE_API_KEY

1. Deploy interceptors/guardrails.py as a Lambda (the deploy script packages it with the SDK).
2. Grant the gateway execution role lambda:InvokeFunction on that function.
3. Register the Lambda on the gateway as a REQUEST and RESPONSE interceptor with passRequestHeaders.
4. Every interceptor reply carries interceptorOutputVersion: "1.0" with the transformed request or response body.

Example

guardrails.py (activefence_client_sdk: src/integration_examples/aws_agentcore_gateway)

import json
import logging
import os
import uuid
from typing import Any

from activefence_client_sdk.client import ActiveFenceClient
from activefence_client_sdk.models import AnalysisContext

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Lazy initialization of ActiveFence client
_af_client: ActiveFenceClient | None = None


def get_af_client() -> ActiveFenceClient:
    """Get or create ActiveFence client (lazy initialization)."""
    global _af_client  # noqa: PLW0603 - one client per warm Lambda container
    if _af_client is None:
        api_key = os.getenv("ALICE_API_KEY")
        if not api_key:
            logger.warning("ALICE_API_KEY not set, Alice evaluation will fail")
        _af_client = ActiveFenceClient(provider="nova", api_key=api_key)
    return _af_client


def _create_analysis_context(user_id: str | None = None, session_id: str | None = None) -> AnalysisContext:
    """Create analysis context for ActiveFence evaluation."""
    if user_id is None:
        user_id = str(uuid.uuid4())
    if session_id is None:
        session_id = str(uuid.uuid4())

    return AnalysisContext(
        session_id=session_id,
        user_id=user_id,
    )


def mcp_request_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """
    Handler for MCP (Model Context Protocol) request interceptors.
    Expected event structure:
    {
        "interceptorInputVersion": "1.0",
        "mcp": {
            "gatewayRequest": {
                "body": {...} or "...",
                ...
            },
            "gatewayResponse": null
        }
    }

    Returns:
    {
        "interceptorOutputVersion": "1.0",
        "mcp": {
            "transformedGatewayRequest": {
                "body": {...}
            }
        }
    }
    """
    logger.info("MCP Request Interceptor received event")

    mcp_data = event.get("mcp", {})
    gateway_request = mcp_data.get("gatewayRequest", {})
    request_body = gateway_request.get("body", {})

    # For most MCP methods (like initialize, tools/list, etc.), we just pass through
    # We could add logic here to evaluate specific methods that contain user input
    # For now, we'll pass through all requests unchanged

    # Return in the required format
    return {
        "interceptorOutputVersion": "1.0",
        "mcp": {"transformedGatewayRequest": {"body": request_body}},
    }


def mcp_response_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """
    Handler for MCP (Model Context Protocol) response interceptors.
    Expected event structure:
    {
        "interceptorInputVersion": "1.0",
        "mcp": {
            "gatewayRequest": {...},
            "gatewayResponse": {
                "body": {...} or "...",
                "statusCode": 200,
                ...
            }
        }
    }

    Returns:
    {
        "interceptorOutputVersion": "1.0",
        "mcp": {
            "transformedGatewayResponse": {
                "body": {...},
                "statusCode": 200
            }
        }
    }
    """
    logger.info("MCP Response Interceptor received event")

    mcp_data = event.get("mcp", {})
    gateway_response = mcp_data.get("gatewayResponse", {})
    response_body = gateway_response.get("body", {})
    status_code = gateway_response.get("statusCode", 200)

    # For now, we'll pass through all responses unchanged
    # In the future, we could evaluate response content for safety

    # Return in the required format
    return {
        "interceptorOutputVersion": "1.0",
        "mcp": {"transformedGatewayResponse": {"body": response_body, "statusCode": status_code}},
    }


def _body_size(payload: Any) -> int:
    """Best-effort size of a gateway request/response body for log metadata."""
    body = payload.get("body") if isinstance(payload, dict) else None
    if body is None:
        return 0
    if isinstance(body, (bytes, bytearray)):
        return len(body)
    if isinstance(body, str):
        return len(body)
    try:
        return len(json.dumps(body))
    except (TypeError, ValueError):
        return -1


def lambda_handler(event, context):
    interception_point = event.get("context", {}).get("interceptionPoint")
    logger.info(
        "Lambda handler invoked (interception_point=%s)",
        interception_point,
    )

    mcp_data = event.get("mcp", {})
    if not mcp_data:
        logger.warning("No mcp data found in event")
        return {"interceptorOutputVersion": "1.0"}

    req = mcp_data.get("gatewayRequest", {})
    res = mcp_data.get("gatewayResponse", {})

    if res:  # response contains the request as well, so we need to check if response is not empty first
        logger.info(
            "RESPONSE intercepted (status=%s, body_size=%d)",
            res.get("statusCode"),
            _body_size(res),
        )
        result = {
            "interceptorOutputVersion": "1.0",
            "mcp": {
                "transformedGatewayResponse": {
                    "statusCode": res.get("statusCode"),  # not working
                    "headers": res.get("headers", {}),
                    "body": res.get("body", {}) or {},
                }
            },
        }
    else:
        logger.info("REQUEST intercepted (body_size=%d)", _body_size(req))
        result = {
            "interceptorOutputVersion": "1.0",
            "mcp": {
                "transformedGatewayRequest": {
                    "headers": req.get("headers", {}),
                    "body": req.get("body", {}) or {},
                }
            },
        }
    return result

deploy the Lambda

python scripts/deploy_lambda.py --profile <aws-profile>   # packages guardrails.py + deps, creates or updates the function, prints the ARN

grant the gateway role invoke permission (add_gateway_lambda_permission.py, identifiers replaced)

policy_document = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "lambda:InvokeFunction",
            "Resource": "<LAMBDA_ARN>",
        }
    ],
}
iam_client.put_role_policy(
    RoleName="<GATEWAY_ROLE_NAME>",
    PolicyName="AllowInvokeGuardrailsInterceptor",
    PolicyDocument=json.dumps(policy_document),
)

register REQUEST + RESPONSE interceptors (register_interceptors.py, identifiers replaced)

response = client.update_gateway(
    gatewayIdentifier="<GATEWAY_ID>",
    name="<GATEWAY_NAME>",
    roleArn="<GATEWAY_ROLE_ARN>",
    protocolType="MCP",
    authorizerType="NONE",
    interceptorConfigurations=[
        {
            "interceptor": {"lambda": {"arn": "<LAMBDA_ARN>"}},
            "interceptionPoints": ["REQUEST", "RESPONSE"],
            "inputConfiguration": {"passRequestHeaders": True},
        }
    ],
)

Good to know

The request and response handlers currently pass the body through unchanged; the evaluate call on tools/call arguments and results is the next step. Identifiers in the register and permission snippets are placeholders.

Links

• AgentCore gateway interceptors

---