AWS Builder Workshop

Build with AgentCore & Strands: Gateway MCP Tool Fabric Developer Workshop

Workshop Summary: Developers build a managed Gateway MCP tool fabric that connects local MCP prototyping, Lambda-backed tools, semantic search, and Strands agent consumption. The workshop walks through schema design, local discovery tests, cloud target registration, JSON-RPC debugging, and SigV4 transport integration. Teams finish with a scalable tool-governance pattern for discoverable, secure, evidence-oriented agent tooling on AWS across complex multi-team platform engineering environments.

Build with AgentCore & Strands: Gateway MCP Tool Fabric Developer Workshop

Audience: backend developers, AI platform developers, AWS integration engineers Duration: 2 hours Primary AWS AI services: Amazon Bedrock AgentCore Gateway, Strands Agents, MCP, AWS Lambda Project output: a managed MCP tool fabric with local MCP tools, Lambda-backed Gateway target, semantic search, and Strands agent invocation.

Educational engineering workshop only. The sample sovereign-risk domain is used to teach agent-tool architecture and is not financial advice.

Workshop Summary

Developers build a managed Gateway MCP tool fabric that connects local MCP prototyping, Lambda-backed tools, semantic search, and Strands agent consumption. The workshop walks through schema design, local discovery tests, cloud target registration, JSON-RPC debugging, and SigV4 transport integration. Teams finish with a scalable tool-governance pattern for discoverable, secure, evidence-oriented agent tooling on AWS across complex multi-team platform engineering environments.


1. Developer learning objectives

Developers will learn how to:

  1. Design agent-facing tool names and schemas.
  2. Build a local streamable HTTP MCP tool server.
  3. Test MCP tool discovery before cloud deployment.
  4. Implement Lambda-backed tool business logic.
  5. Register Lambda tools in Amazon Bedrock AgentCore Gateway.
  6. Enable semantic tool search.
  7. Call Gateway tools from a Strands agent with SigV4 transport.
  8. Add direct JSON-RPC tool-call tests for debugging.

2. Architecture built in this workshop

Local developer environment
  ├─ mcp_server.py               # local MCP server
  ├─ mcp_client_local.py         # local discovery client
  ├─ lambda_function.py          # Lambda target logic
  ├─ package_lambda.sh           # Lambda zip packaging
  ├─ gateway_setup.py            # Gateway and target creation
  ├─ semantic_search.py          # Gateway semantic search call
  ├─ direct_tool_call.py         # JSON-RPC tools/call debugging
  ├─ strands_gateway_agent.py    # Strands agent using Gateway MCP tools
  └─ prompts/tool_selection.md
AWS
  ├─ Amazon Bedrock AgentCore Gateway
  ├─ AWS Lambda target
  ├─ Cognito / custom JWT authorizer inputs
  ├─ IAM Gateway role
  └─ Strands agent client with SigV4 MCP transport

3. 2-hour hands-on agenda

TimeModuleHands-on output
0–10Tool-fabric conceptsMCP, Gateway, Lambda target flow understood
10–25Tool contractTool names, descriptions, and schemas planned
25–45Local MCP serverStreamable HTTP tool server running
45–60Local discoveryTool list and metadata verified
60–80Lambda targetTool backend packaged and deployed or prepared
80–100Gateway targetMCP Gateway and Lambda target created
100–110Semantic searchSearch tool returns relevant tools
110–120Strands agentAgent calls Gateway tools

Step 1 — Define the tool-selection prompt and schema policy

Developer action

mkdir -p agentcore-strands-gateway/prompts
cd agentcore-strands-gateway
cat > prompts/tool_selection.md <<'EOF'
You are a tool-selection planner for sovereign risk workflows.
Use semantic search when the tool inventory is large.
Select the smallest set of tools that can answer the user's request.
Return selected tool names, arguments, expected evidence, and reason for selection.
Do not call execution or trading tools. This system is for engineering analysis only.
EOF

Create tool_schema.py.

TOOL_SCHEMA = [
    {
        "name": "funding_liquidity",
        "description": "Assess repo stress, money-market pressure, USD funding, and cash preference.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Funding liquidity analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "credit_spread_widening",
        "description": "Assess IG/HY spread widening, CDS pressure, downgrade risk, and liquidity premium.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Credit spread analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "sovereign_debt_risk_repricing",
        "description": "Assess fiscal credibility, real yields, policy divergence, capital flows, and FX pressure.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Sovereign repricing analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "currency_mismatch",
        "description": "Assess FX mismatch, external debt pressure, reserves, basis swaps, and hedge context.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Currency mismatch request."}},
            "required": ["query"],
        },
    },
]

Business logic

The schema defines the tool vocabulary exposed to agents. It also defines the user intent each tool is responsible for.

Code logic

Each schema item contains a tool name, description, JSON input schema, and required fields. The same schema is reused by Gateway setup.

Expected result

Developers have one canonical tool schema source instead of duplicating schemas in multiple scripts.

System design decision

  • Canonical schema source: Tool schemas influence discovery, validation, and agent behavior. Keeping them in tool_schema.py prevents drift between local tests, Gateway registration, and documentation. This is important as tool catalogs grow.
  • Descriptions optimized for search: Descriptions include business terms such as repo stress, CDS pressure, real yields, and FX pressure. Semantic search depends on meaningful metadata, so descriptions should be treated as production-quality API documentation.
  • Smallest tool set principle: The prompt tells agents to select the smallest sufficient tool set. This reduces unnecessary tool calls, lowers latency, and makes audit trails easier to review.

Step 2 — Build a local MCP server

Developer action

Create mcp_server.py.

from mcp.server.fastmcp import FastMCP
mcp = FastMCP(host="0.0.0.0", port=8000, stateless_http=True)
@mcp.tool()
def funding_liquidity(query: str) -> dict:
    """Assess repo stress, money-market pressure, USD funding, and cash preference."""
    return {
        "tool": "funding_liquidity",
        "query": query,
        "signals": ["repo rates", "swap spreads", "USD funding", "cash preference"],
        "summary": "Funding stress should be checked before interpreting spread movement as pure credit risk.",
    }
@mcp.tool()
def credit_spread_widening(query: str) -> dict:
    """Assess IG/HY spread widening, CDS pressure, downgrade risk, and liquidity premium."""
    return {
        "tool": "credit_spread_widening",
        "query": query,
        "signals": ["IG spreads", "HY spreads", "CDS index", "downgrade watch"],
        "summary": "Credit spread widening may reflect default-risk repricing and liquidity withdrawal.",
    }
@mcp.tool()
def sovereign_debt_risk_repricing(query: str) -> dict:
    """Assess fiscal credibility, real yields, policy divergence, capital flows, and FX pressure."""
    return {
        "tool": "sovereign_debt_risk_repricing",
        "query": query,
        "signals": ["real yields", "fiscal path", "central-bank reaction", "capital flows"],
        "summary": "Sovereign repricing should separate fiscal risk, policy divergence, and capital-flow pressure.",
    }
@mcp.tool()
def currency_mismatch(query: str) -> dict:
    """Assess FX mismatch, external debt pressure, reserves, basis swaps, and hedge context."""
    return {
        "tool": "currency_mismatch",
        "query": query,
        "signals": ["FX reserves", "external debt", "basis swaps", "forward points"],
        "summary": "Currency mismatch can amplify sovereign stress when external funding tightens.",
    }
if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Run:

python mcp_server.py

Business logic

Each MCP tool returns structured evidence about one risk dimension.

Code logic

FastMCP exposes Python functions as tools over streamable HTTP. Each function accepts a query and returns a dictionary.

Expected result

A local MCP endpoint is available at http://localhost:8000/mcp.

System design decision

  • Structured tool output: Returning dictionaries instead of plain strings gives agents and tests inspectable fields: tool name, query, signals, and summary. This improves grounding and future evaluation.
  • Local server before Gateway: Local MCP testing reduces cloud complexity. Developers first verify tool behavior and metadata, then publish tools behind AgentCore Gateway.
  • Stateless tool design: Tools do not depend on hidden server memory. This makes them easier to scale, test, and deploy behind managed infrastructure.

Step 3 — Test local MCP discovery and direct calls

Developer action

Create mcp_client_local.py.

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def main():
    async with streamablehttp_client(
        "http://localhost:8000/mcp",
        {},
        timeout=120,
        terminate_on_close=False,
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("Discovered tools:")
            for tool in tools.tools:
                print("-", tool.name, "|", tool.description)
            result = await session.call_tool(
                "sovereign_debt_risk_repricing",
                {"query": "US-China yield spread widening and RMB pressure"},
            )
            print("Direct call result:")
            print(result)
if __name__ == "__main__":
    asyncio.run(main())

Business logic

The client validates both discovery and direct execution before Gateway is introduced.

Code logic

The client initializes an MCP session, lists tools, then calls one tool with arguments.

Expected result

The console prints tool metadata and a structured tool result.

System design decision

  • Discovery plus direct-call test: Listing tools only proves metadata visibility. Direct calls prove the tool accepts arguments and returns a valid response. Both are required before cloud publication.
  • No model in the loop: The test isolates MCP transport and tool behavior from LLM reasoning. This makes failures easier to debug.
  • Representative query: The direct call uses a realistic query containing yield spread and RMB pressure. Testing with domain language helps validate descriptions and output usefulness.

Step 4 — Implement Lambda target logic

Developer action

Create lambda_function.py.

import json
HANDLERS = {
    "funding_liquidity": {
        "signals": ["repo stress", "money-market spreads", "USD funding", "swap basis"],
        "summary": "Check funding stress before treating yield movement as isolated sovereign repricing.",
    },
    "credit_spread_widening": {
        "signals": ["IG spreads", "HY spreads", "CDS indices", "downgrade risk"],
        "summary": "Credit spread widening can indicate default-risk repricing or liquidity premium expansion.",
    },
    "sovereign_debt_risk_repricing": {
        "signals": ["real yields", "fiscal credibility", "policy divergence", "capital flows"],
        "summary": "Sovereign repricing should be decomposed into fiscal, policy, flow, and FX drivers.",
    },
    "currency_mismatch": {
        "signals": ["external debt", "FX reserves", "basis swaps", "forward hedging cost"],
        "summary": "Currency mismatch can amplify stress when external refinancing conditions tighten.",
    },
}
def lambda_handler(event, context):
    tool_name = event.get("toolName") or event.get("name")
    arguments = event.get("arguments", {})
    query = arguments.get("query", "")
    if tool_name not in HANDLERS:
        return {
            "statusCode": 404,
            "body": json.dumps({"error": f"Unknown tool: {tool_name}"}),
        }
    output = {
        "tool": tool_name,
        "query": query,
        "signals": HANDLERS[tool_name]["signals"],
        "summary": HANDLERS[tool_name]["summary"],
    }
    return {"statusCode": 200, "body": json.dumps(output)}

Create packaging helper:

cat > package_lambda.sh <<'EOF'
#!/usr/bin/env bash
set -euo pipefail
rm -f lambda_function.zip
zip lambda_function.zip lambda_function.py
ls -lh lambda_function.zip
EOF
chmod +x package_lambda.sh
./package_lambda.sh

Business logic

Lambda becomes the governed backend for Gateway tool calls. It maps tool requests to structured signal outputs.

Code logic

The handler reads toolName and arguments, validates the tool, and returns JSON. The packaging script creates a zip artifact.

Expected result

lambda_function.zip is ready for deployment or for a prepared workshop bootstrap script.

System design decision

  • Lambda as backend boundary: Gateway exposes tools, while Lambda owns deterministic business logic. This separates model reasoning from backend execution and makes tool behavior testable and observable.
  • Shared handler map: A dictionary-based router keeps the workshop compact while still demonstrating how multiple tools can map to one Lambda target. Production systems can split handlers later.
  • Structured response body: Returning tool, query, signals, and summary gives the agent evidence it can cite before synthesis. This also improves auditability.

Step 5 — Create AgentCore Gateway and register target

Developer action

Create gateway_setup.py.

import os
import boto3
from tool_schema import TOOL_SCHEMA
region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
client = boto3.client("bedrock-agentcore-control", region_name=region)
gateway = client.create_gateway(
    name="gateway-sovereign-risk-tools-hands-on",
    roleArn=os.environ["AGENTCORE_GATEWAY_ROLE_ARN"],
    protocolType="MCP",
    authorizerType="CUSTOM_JWT",
    authorizerConfiguration={
        "customJWTAuthorizer": {
            "allowedClients": [os.environ["COGNITO_CLIENT_ID"]],
            "discoveryUrl": os.environ["COGNITO_DISCOVERY_URL"],
        }
    },
    protocolConfiguration={
        "mcp": {
            "searchType": "SEMANTIC",
            "supportedVersions": ["2025-03-26"],
        }
    },
    description="Hands-on MCP Gateway for sovereign risk tools",
)
target = client.create_gateway_target(
    gatewayIdentifier=gateway["gatewayId"],
    name="sovereign-risk-lambda-target",
    description="Lambda target exposing sovereign risk tools",
    targetConfiguration={
        "mcp": {
            "lambda": {
                "lambdaArn": os.environ["SOVEREIGN_TOOLS_LAMBDA_ARN"],
                "toolSchema": {"inlinePayload": TOOL_SCHEMA},
            }
        }
    },
    credentialProviderConfigurations=[{"credentialProviderType": "GATEWAY_IAM_ROLE"}],
)
print("Gateway ID:", gateway["gatewayId"])
print("Gateway URL:", gateway["gatewayUrl"])
print("Target ID:", target["targetId"])

Business logic

Gateway publishes Lambda-backed sovereign-risk tools through a managed MCP endpoint.

Code logic

The script creates the Gateway, configures JWT authorization, enables semantic search, and registers the Lambda target using the canonical schema.

Expected result

Developers receive a Gateway URL and target ID.

System design decision

  • Gateway centralizes tool access: AgentCore Gateway becomes the managed MCP endpoint for agents. This avoids every agent directly integrating with Lambda, auth, and transport logic.
  • JWT ingress with IAM egress: Inbound callers authenticate with JWT, while Gateway invokes Lambda using IAM. This separates caller authorization from backend execution credentials.
  • Semantic discovery enabled: The gateway indexes tool metadata for semantic search. This prepares the system for larger tool catalogs without prompt-bloating every schema into the agent context.

Step 6 — Test Gateway semantic search and direct JSON-RPC calls

Developer action

Create semantic_search.py.

import json
import os
import requests
def call_gateway(payload):
    response = requests.post(
        os.environ["AGENTCORE_GATEWAY_URL"],
        json=payload,
        headers={
            "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
            "Content-Type": "application/json",
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()
payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "x_amz_bedrock_agentcore_search",
        "arguments": {"query": "RMB pressure and sovereign debt repricing"},
    },
}
print(json.dumps(call_gateway(payload), indent=2))

Create direct_tool_call.py.

import json
import os
import requests
payload = {
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
    "params": {
        "name": "sovereign_debt_risk_repricing",
        "arguments": {"query": "US-China yield spread widening and capital-flow pressure"},
    },
}
response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)
print(json.dumps(response.json(), indent=2))

Business logic

Semantic search finds relevant tools by intent. Direct calls verify a selected tool executes through Gateway.

Code logic

Both scripts send JSON-RPC tools/call requests. One calls the built-in search tool; the other calls a domain tool.

Expected result

Search returns ranked tools, and direct call returns Lambda-backed structured evidence.

System design decision

  • Search and direct execution are separate tests: Semantic search validates discovery. Direct call validates execution. Keeping them separate helps isolate failures in auth, schema, routing, or Lambda logic.
  • JSON-RPC debugging path: Direct HTTP calls are useful when an agent behaves unexpectedly. Developers can reproduce the exact Gateway request without model reasoning in the loop.
  • Bearer token boundary: Gateway calls include authorization headers. This reinforces that tool discovery and execution are protected operations, not anonymous APIs.

Step 7 — Call Gateway from a Strands agent

Developer action

Create strands_gateway_agent.py.

import os
import boto3
from botocore.credentials import Credentials
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from streamable_http_sigv4 import streamablehttp_client_with_sigv4
SERVICE = "bedrock-agentcore"
def assume_gateway_role(role_arn: str):
    return boto3.client("sts").assume_role(
        RoleArn=role_arn,
        RoleSessionName="strands-gateway-agent-hands-on",
        DurationSeconds=3600,
    )["Credentials"]
def main():
    region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
    creds = assume_gateway_role(os.environ["AGENTCORE_GATEWAY_INVOKE_ROLE_ARN"])
    mcp_client = MCPClient(lambda: streamablehttp_client_with_sigv4(
        url=os.environ["AGENTCORE_GATEWAY_URL"],
        credentials=Credentials(
            creds["AccessKeyId"],
            creds["SecretAccessKey"],
            creds["SessionToken"],
        ),
        service=SERVICE,
        region=region,
    ))
    with mcp_client:
        tools = mcp_client.list_tools_sync()
        print("Tools loaded:", [tool.tool_name for tool in tools])
        agent = Agent(
            model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2),
            tools=tools,
            system_prompt=(
                "You are a tool-using risk engineering assistant. "
                "Use Gateway tools for evidence first. Then synthesize limitations, confirmation signals, and invalidation triggers. "
                "Do not provide investment advice or autonomous trading instructions."
            ),
        )
        result = agent(
            "Analyze RMB pressure, sovereign debt repricing, and credit spread widening from wider US-China yield spreads."
        )
        print(result)
if __name__ == "__main__":
    main()

Business logic

The Strands agent uses Gateway tools as evidence sources before writing a final synthesis.

Code logic

The script assumes an IAM role, creates SigV4 MCP transport, lists Gateway tools, loads them into Strands, and invokes the agent.

Expected result

The agent prints loaded tools and returns a tool-grounded analysis.

System design decision

  • Strands consumes managed MCP tools: The agent does not call Lambda directly. Gateway owns tool exposure, auth, and translation. This makes the agent independent of backend implementation details.
  • Temporary credentials: The agent uses assumed-role credentials, reducing long-lived secret exposure. IAM defines what Gateway access the agent has.
  • Evidence-first agent prompt: The system prompt asks for tool evidence before synthesis. This improves auditability and makes output easier to evaluate.

Final developer checklist

  • <input type="checkbox" disabled> Local MCP server runs.
  • <input type="checkbox" disabled> Local MCP client lists and calls tools.
  • <input type="checkbox" disabled> Lambda zip package exists.
  • <input type="checkbox" disabled> Gateway and target are created.
  • <input type="checkbox" disabled> Semantic search returns relevant tools.
  • <input type="checkbox" disabled> Direct JSON-RPC tool call succeeds.
  • <input type="checkbox" disabled> Strands agent loads Gateway tools and produces grounded output.

Additional Hands-on Developer Labs

These labs extend the Gateway MCP Tool Fabric workshop with deeper debugging, schema governance, authentication practice, and agent-tool evaluation. They are intentionally different from the core build and focus on making tool ecosystems production-ready.


Hands-on Lab A — Add schema linting for MCP tool definitions

Developer goal

Validate tool schemas before registering them in AgentCore Gateway.

Developer action

Create lint_tool_schema.py:

from tool_schema import TOOL_SCHEMA
REQUIRED_TOP_LEVEL = {"name", "description", "inputSchema"}
def lint_tool_schema(schema: list[dict]) -> list[str]:
    errors = []
    names = set()
    for index, tool in enumerate(schema):
        missing = REQUIRED_TOP_LEVEL - set(tool.keys())
        if missing:
            errors.append(f"Tool index {index} missing keys: {sorted(missing)}")
        name = tool.get("name")
        if not name:
            errors.append(f"Tool index {index} has empty name")
        elif name in names:
            errors.append(f"Duplicate tool name: {name}")
        names.add(name)
        description = tool.get("description", "")
        if len(description.split()) < 6:
            errors.append(f"Tool {name} description is too short for semantic search")
        input_schema = tool.get("inputSchema", {})
        if input_schema.get("type") != "object":
            errors.append(f"Tool {name} inputSchema must be object")
        if "required" not in input_schema:
            errors.append(f"Tool {name} inputSchema should define required fields")
    return errors
if __name__ == "__main__":
    errors = lint_tool_schema(TOOL_SCHEMA)
    if errors:
        print("Schema lint failed")
        print("\n".join(errors))
        raise SystemExit(1)
    print(f"Validated {len(TOOL_SCHEMA)} tool schemas")

Run:

python lint_tool_schema.py

Business logic

Tool schemas are part of the agent API. Bad schemas reduce discovery quality and can cause runtime invocation failures.

Code logic

The linter checks required fields, duplicate tool names, description quality, object schemas, and required input fields.

Expected result

Validated 4 tool schemas

System design decision

  • Schema quality before registration: Gateway registration should not be the first time a schema is checked. Local linting catches simple problems such as duplicate names and weak descriptions before cloud resources are changed.
  • Semantic search depends on descriptions: A short or vague description can make the search index less useful. The linter enforces minimum description quality because metadata is operational behavior for agents.
  • Fast CI-friendly check: The script exits with code 1 when validation fails, making it easy to run in CI pipelines or pre-commit hooks.

Hands-on Lab B — Build a direct MCP tools/list tester for Gateway

Developer goal

Debug Gateway tool discovery with raw JSON-RPC before involving Strands.

Developer action

Create gateway_list_tools.py:

import json
import os
import requests
payload = {
    "jsonrpc": "2.0",
    "id": 100,
    "method": "tools/list",
    "params": {},
}
response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)
print("HTTP", response.status_code)
print(json.dumps(response.json(), indent=2))

Business logic

Tool discovery must work before agents can call tools. This lab verifies Gateway discovery directly.

Code logic

The script sends a JSON-RPC tools/list request to the Gateway endpoint with a bearer token.

Expected result

The response contains the registered MCP tools and their schemas.

System design decision

  • Raw protocol debugging: When Strands agent behavior is confusing, developers need a lower-level test. JSON-RPC calls isolate Gateway discovery from model reasoning and Strands orchestration.
  • Authentication visibility: The script makes the bearer token requirement explicit. This helps developers distinguish auth failures from tool registration failures.
  • Operational smoke test: tools/list can become a deployment smoke test to confirm that new Gateway targets are visible after registration.

Hands-on Lab C — Add a Gateway tool-call replay file

Developer goal

Create repeatable JSON-RPC test payloads that can be replayed during debugging.

Developer action

Create requests/sovereign_repricing_call.json:

{
  "jsonrpc": "2.0",
  "id": 201,
  "method": "tools/call",
  "params": {
    "name": "sovereign_debt_risk_repricing",
    "arguments": {
      "query": "US-China yield spread widening, RMB pressure, and capital-flow stress"
    }
  }
}

Create replay_gateway_request.py:

import json
import os
import sys
import requests
path = sys.argv[1]
payload = json.load(open(path, encoding="utf-8"))
response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)
print(json.dumps(response.json(), indent=2))

Run:

mkdir -p requests
python replay_gateway_request.py requests/sovereign_repricing_call.json

Business logic

Replay files create repeatable evidence for debugging tool behavior and demonstrating expected Gateway responses.

Code logic

The replay script loads a JSON payload from disk and posts it to Gateway.

Expected result

The same tool call can be replayed repeatedly without rewriting Python code.

System design decision

  • Reproducible tool debugging: Saved JSON-RPC payloads make it easy to reproduce bugs and compare responses across deployments. This is valuable when tool schemas or Lambda logic change.
  • Separation of request and runner: The request file is data, while the Python script is a generic runner. This allows many test cases to reuse one client.
  • Useful for documentation: Replay files double as examples for other developers who need to understand how to call Gateway tools directly.

Hands-on Lab D — Add tool-result normalization for Strands responses

Developer goal

Normalize tool outputs before the agent synthesizes a final answer.

Developer action

Create tool_result_normalizer.py:

import json
def normalize_tool_result(raw_result) -> dict:
    if isinstance(raw_result, dict):
        return raw_result
    if isinstance(raw_result, str):
        try:
            return json.loads(raw_result)
        except json.JSONDecodeError:
            return {"raw_text": raw_result}
    if isinstance(raw_result, list):
        return {"items": raw_result}
    return {"repr": repr(raw_result)}

Use it in debugging code:

from tool_result_normalizer import normalize_tool_result
normalized = normalize_tool_result(result)
print(normalized)

Business logic

Agents may receive tool outputs in different shapes. Normalization makes downstream evidence handling predictable.

Code logic

The helper converts dictionaries, JSON strings, lists, and unknown objects into a dictionary shape.

Expected result

Tool output can be logged, evaluated, and rendered consistently.

System design decision

  • Predictable evidence format: Tool outputs should be machine-readable where possible. Normalization lets clients and evaluations inspect fields without depending on every backend returning the exact same shape.
  • Defensive integration: Gateway targets may evolve or return unexpected payloads. Normalization prevents brittle agent wrappers from failing on small output-shape changes.
  • Evaluation readiness: Consistent tool evidence makes it easier to check whether the final agent response used the expected tools and signals.

Hands-on Lab E — Add semantic search comparison tests

Developer goal

Compare semantic search results for different user intents and verify the expected tool appears in the top results.

Developer action

Create semantic_search_eval.py:

import json
import os
import requests
CASES = [
    {"query": "repo funding pressure and cash preference", "expected": "funding_liquidity"},
    {"query": "CDS widening and downgrade risk", "expected": "credit_spread_widening"},
    {"query": "fiscal credibility and real-yield repricing", "expected": "sovereign_debt_risk_repricing"},
    {"query": "FX reserves and external dollar debt", "expected": "currency_mismatch"},
]
def search(query: str) -> list[str]:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "x_amz_bedrock_agentcore_search",
            "arguments": {"query": query},
        },
    }
    response = requests.post(
        os.environ["AGENTCORE_GATEWAY_URL"],
        json=payload,
        headers={"Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}"},
        timeout=30,
    ).json()
    tools = response.get("result", {}).get("structuredContent", {}).get("tools", [])
    return [tool.get("name") for tool in tools]
for case in CASES:
    names = search(case["query"])
    passed = case["expected"] in names[:3]
    print(case["query"], "PASS" if passed else "FAIL", names[:3])

Business logic

Semantic search quality determines whether agents find the right tools for business-language requests.

Code logic

The script runs multiple search queries and checks whether expected tools appear in the top three results.

Expected result

Each test prints PASS with top matching tools.

System design decision

  • Search quality as testable behavior: Semantic search should be validated like any other feature. If descriptions or schemas change, search quality can regress.
  • Intent-based test cases: The cases use business language rather than exact tool names. This tests whether metadata supports real user prompts.
  • Top-k tolerance: Checking the top three results is more realistic than requiring the first result every time. Semantic retrieval may rank close tools differently, but the expected tool should still be discoverable.

Hands-on Lab F — Add least-privilege environment checklist

Developer goal

Create an operations checklist for the Gateway tool fabric before production promotion.

Developer action

Create docs/gateway_operational_checklist.md:

# Gateway Operational Checklist
## Identity and authorization
- [ ] Gateway uses CUSTOM_JWT authorizer.
- [ ] Allowed clients are restricted to approved application clients.
- [ ] Token lifetime and refresh process are documented.
## IAM and target access
- [ ] Gateway role can invoke only approved Lambda targets.
- [ ] Lambda resource policy does not allow broad public invocation.
- [ ] Developer credentials are not embedded in code.
## Tool governance
- [ ] Tool schemas are linted before registration.
- [ ] Tool descriptions are long enough for semantic search.
- [ ] Tool outputs are structured and auditable.
- [ ] Dangerous or execution-style tools are not registered in this analysis gateway.
## Observability
- [ ] Gateway target invocation errors are monitored.
- [ ] Lambda logs include request IDs or tool names.
- [ ] Semantic search tests pass after schema changes.

Business logic

The checklist helps platform teams review security, governance, and observability before exposing tools broadly.

Code logic

This is Markdown documentation, but it becomes an operational artifact for pull requests and production reviews.

Expected result

Teams have a repeatable readiness checklist for the Gateway tool fabric.

System design decision

  • Documentation as control: Operational checklists prevent important production concerns from being tribal knowledge. They are especially useful when multiple teams publish tools.
  • Least-privilege focus: Gateway centralizes tool access, so IAM and JWT boundaries must be reviewed carefully. The checklist makes those boundaries explicit.
  • Promotion gate: The checklist can become a release requirement before a Gateway target moves from workshop to shared development or production environments.