Build with AgentCore & Strands: Gateway Factory Engineering 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 etch-process-window domain is used to teach agent-tool architecture and is not process-release advice.

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

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

Developer action

mkdir -p agentcore-strands-factory-gateway/prompts
cd agentcore-strands-factory-gateway
cat > prompts/tool_selection.md <<'EOF'
You are a tool-selection planner for etch process window 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 equipment-control tools. This system is for engineering analysis only.
EOF

Create tool_schema.py.

TOOL_SCHEMA = [
    {
        "name": "throughput_throughput",
        "description": "Assess WIP queue stress, tool availability pressure, WAFERS throughput, and hot-lot preference.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Throughput stability analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "metrology_drift_widening",
        "description": "Assess inline/lot-level drift widening, CD-SEM pressure, yield-loss risk, and capacity premium.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Critical-dimension drift analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "process_window_drift",
        "description": "Assess process capability, overlay error, recipe divergence, lot flow, and overlay drift.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Etch Process drift analysis request."}},
            "required": ["query"],
        },
    },
    {
        "name": "tool_to_tool_mismatch",
        "description": "Assess overlay mismatch, backlog pressure, reserves, baseline offsets, and control context.",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "Tool-to-tool 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 WIP queue stress, CD-SEM pressure, overlay error, and overlay drift. 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 throughput_throughput(query: str) -> dict:
    """Assess WIP queue stress, tool availability pressure, WAFERS throughput, and hot-lot preference."""
    return {
        "tool": "throughput_throughput",
        "query": query,
        "signals": ["queue depth", "swap drifts", "WAFERS throughput", "hot-lot preference"],
        "summary": "Throughput stress should be checked before interpreting drift movement as pure metrology risk.",
    }

@mcp.tool()
def metrology_drift_widening(query: str) -> dict:
    """Assess inline/lot-level drift widening, CD-SEM pressure, yield-loss risk, and capacity premium."""
    return {
        "tool": "metrology_drift_widening",
        "query": query,
        "signals": ["inline drifts", "lot-level drifts", "CD-SEM index", "yield-loss watch"],
        "summary": "CD-SEM drift widening may reflect defect-risk drift and capacity withdrawal.",
    }

@mcp.tool()
def process_window_drift(query: str) -> dict:
    """Assess process capability, overlay error, recipe divergence, lot flow, and overlay drift."""
    return {
        "tool": "process_window_drift",
        "query": query,
        "signals": ["overlay error", "fiscal path", "equipment controller reaction", "lot flow"],
        "summary": "Etch Process drift should separate fiscal risk, recipe divergence, and lot-flow pressure.",
    }

@mcp.tool()
def tool_to_tool_mismatch(query: str) -> dict:
    """Assess overlay mismatch, backlog pressure, reserves, baseline offsets, and control context."""
    return {
        "tool": "tool_to_tool_mismatch",
        "query": query,
        "signals": ["overlay reserves", "external process-window", "baseline offsets", "forecast offsets"],
        "summary": "Tool-to-tool mismatch can amplify etch-process stress when external throughput tightens.",
    }

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Run:

python mcp_server.py

Business logic

Each MCP tool returns structured evidence about one process 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(
                "process_window_drift",
                {"query": "Fab-A/Fab-B yield drift widening and overlay drift"},
            )
            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 drift and overlay drift. 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 = {
    "throughput_throughput": {
        "signals": ["WIP queue stress", "tool availability deltas", "WAFERS throughput", "swap basis"],
        "summary": "Check queue stress before treating yield movement as isolated etch-process drift.",
    },
    "metrology_drift_widening": {
        "signals": ["inline drifts", "lot-level drifts", "CD-SEM indices", "yield-loss risk"],
        "summary": "CD-SEM drift widening can indicate defect-risk drift or capacity premium expansion.",
    },
    "process_window_drift": {
        "signals": ["overlay error", "process capability", "recipe divergence", "lot flow"],
        "summary": "Etch Process drift should be decomposed into fiscal, policy, flow, and overlay drivers.",
    },
    "tool_to_tool_mismatch": {
        "signals": ["external process-window", "overlay reserves", "baseline offsets", "forward hedging cost"],
        "summary": "Tool-to-tool 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-etch-process-window-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 etch process window tools",
)

target = client.create_gateway_target(
    gatewayIdentifier=gateway["gatewayId"],
    name="etch-process-window-lambda-target",
    description="Lambda target exposing etch process window tools",
    targetConfiguration={
        "mcp": {
            "lambda": {
                "lambdaArn": os.environ["FACTORY_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 etch-process-window 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": "overlay drift and etch-process process-window drift"},
    },
}

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": "process_window_drift",
        "arguments": {"query": "Fab-A/Fab-B yield drift widening and lot-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",
        Queue-timeSeconds=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 factory engineering automation assistant. "
                "Use Gateway tools for evidence first. Then synthesize limitations, confirmation signals, and invalidation triggers. "
                "Do not provide process-release advice or autonomous equipment-control instructions."
            ),
        )

        result = agent(
            "Analyze overlay drift, etch-process process-window drift, and CD-SEM drift widening from wider Fab-A/Fab-B yield drifts."
        )
        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

• [ ] Local MCP server runs.
• [ ] Local MCP client lists and calls tools.
• [ ] Lambda zip package exists.
• [ ] Gateway and target are created.
• [ ] Semantic search returns relevant tools.
• [ ] Direct JSON-RPC tool call succeeds.
• [ ] 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/etch-process_drift_call.json:

{
  "jsonrpc": "2.0",
  "id": 201,
  "method": "tools/call",
  "params": {
    "name": "process_window_drift",
    "arguments": {
      "query": "Fab-A/Fab-B yield drift widening, overlay drift, and lot-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/etch-process_drift_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 throughput pressure and hot-lot preference", "expected": "throughput_throughput"},
    {"query": "CDS widening and yield-loss risk", "expected": "metrology_drift_widening"},
    {"query": "process capability and overlay-error drift", "expected": "process_window_drift"},
    {"query": "overlay reserves and cross-fab WIP", "expected": "tool_to_tool_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.