AgentCore Gateway

uv run bootcamp.py test --only 2

class ReadOnlyGuardHook(HookProvider):
    """Blocks destructive SQL. Replaces Phase 1's interactive approval: a runtime has no human to ask.

    shared/sql_guard.py decides: only SELECT / EXPLAIN / read-only WITH queries pass.
    """

    def register_hooks(self, registry: HookRegistry, **kwargs) -> None:
        """Run `guard` before every tool call."""
        registry.add_callback(BeforeToolCallEvent, self.guard)

    def guard(self, event: BeforeToolCallEvent) -> None:
        """Cancel any `query_db` call (direct or via the gateway target) that may write."""
        if not event.tool_use.get("name", "").endswith("query_db"):
            return
        query = str(event.tool_use.get("input", {}).get("query", ""))
        if sql_guard.is_write(query):
            event.cancel_tool = "Destructive SQL requires human approval and is blocked in production."

TOOL_NAME_KEY = "bedrockAgentCoreToolName"


def get_us_forecast(arguments: dict[str, Any]) -> str:
    """The next few NWS forecast periods for a US latitude/longitude (compact text, US-only reply elsewhere)."""
    return nws_weather.get_us_forecast(float(arguments["latitude"]), float(arguments["longitude"]))


TOOLS: dict[str, Callable[[dict[str, Any]], str]] = {"get_us_forecast": get_us_forecast}
"""Tool name (as in tools.json) -> implementation."""


def tool_name(context: Any) -> str:
    """The bare tool name the Gateway called (`Weather___get_us_forecast` -> `get_us_forecast`), or ""."""
    custom = getattr(getattr(context, "client_context", None), "custom", None) or {}
    return str(custom.get(TOOL_NAME_KEY, "")).split(TARGET_SEPARATOR)[-1]


def mcp_text(text: str, is_error: bool = False) -> dict[str, Any]:
    """An MCP tool result. The Gateway passes `content` through as is; a bare string would reach the model
    JSON-quoted, with every newline escaped."""
    return {"content": [{"type": "text", "text": text}], "isError": is_error}


def handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """Lambda entry point: run the requested tool; problems come back as an `Error: ...` text the model can read.

    Args:
        event: The tool's arguments, as the model sent them through the Gateway.
        context: The Lambda context; carries the Gateway's tool name.

    Returns:
        The tool's answer as an MCP tool result.
    """
    name = tool_name(context)
    tool = TOOLS.get(name)
    if tool is None:
        return mcp_text(f"Error: unknown tool {name!r}; this target serves {sorted(TOOLS)}.", is_error=True)
    try:
        return mcp_text(tool(event))
    except (KeyError, TypeError, ValueError) as error:
        return mcp_text(f"Error: bad arguments for {name} ({type(error).__name__}: {error}).", is_error=True)

[
  {
    "name": "get_us_forecast",
    "description": "Get the National Weather Service forecast (next few periods: temperature, wind, conditions) for a location in the United States. US only.",
    "inputSchema": {
      "type": "object",
      "properties": {
        "latitude": {
          "type": "number",
          "description": "Latitude of the US location in decimal degrees, e.g. 47.6062 for Seattle."
        },
        "longitude": {
          "type": "number",
          "description": "Longitude of the US location in decimal degrees, e.g. -122.3321 for Seattle."
        }
      },
      "required": ["latitude", "longitude"]
    }
  }
]

SEARCH_TOOL = "x_amz_bedrock_agentcore_search"
"""The Gateway's built-in semantic tool search (`search_type = "SEMANTIC"`, terraform/participant/task2_gateway.tf)."""


DATA_TOOLS = ("Run read-only SQL on the DataStream company database: employees, departments, projects, tables", 3)


WEATHER_TOOLS = ("Weather forecast for a location in the United States", 1)
"""(what the specialist asks the Gateway's tool search for, how many of the best-ranked tools it keeps).

Search ranks every tool, best match first, and has no cut-off: the specialist decides how many it needs (query_db
plus the list_tables/describe_table challenge tools; one weather tool)."""


def search_tools(gateway: MCPClient, need: tuple[str, int]) -> list[MCPAgentTool]:
    """The best-ranked Gateway tools for `need` (query, how many), found by the Gateway's semantic search.

    Only these tools' schemas go into the specialist's context, not every tool behind the Gateway: each schema costs
    input tokens on every model call, and a long tool list makes the model pick the wrong tool more often.
    """
    query, top_k = need
    result = gateway.call_tool_sync(f"search-{uuid.uuid4().hex[:12]}", SEARCH_TOOL, {"query": query})
    if result.get("status") != "success":
        raise RuntimeError(f"Gateway tool search failed: {result.get('content')}")
    found = json.loads("".join(part.get("text", "") for part in result.get("content", []))).get("tools", [])
    ranked = [spec for spec in found if spec.get("name") != SEARCH_TOOL][:top_k]
    return [MCPAgentTool(McpTool.model_validate(spec), gateway) for spec in ranked]

uv run bootcamp.py tools
uv run bootcamp.py tools --search "employee headcount by department"

$ uv run bootcamp.py tools
Gateway tools (3, plus the built-in x_amz_bedrock_agentcore_search):
  DataStreamDatabase: query_db
  Weather: get_us_alerts, get_us_forecast
Semantic search "weather forecast for a US city" -> Weather___get_us_forecast
Prompt tokens for "What will the weather be like in Seattle tomorrow?" on gpt-6-luna:
  no tools               16
  every tool (3)        212  (+196, ~65 per tool)
  search, top 1         123  (+107)
At 200 tools, loading every tool would add ~13,067 tokens to every model call; with search the agent pays only for the few tools that match.

$ uv run bootcamp.py tools --search "employee headcount by department"
Semantic search "employee headcount by department" -> Weather___get_us_alerts
  ...

uv run bootcamp.py deploy
uv run bootcamp.py test --only 2

[CHALLENGE] SKIP stage 2 describe_table via the Gateway: no DataStreamDatabase___describe_table on your Gateway yet
# after your change:
[CHALLENGE] PASS stage 2 describe_table via the Gateway: DataStreamDatabase___describe_table(employees) -> 14 rows, updated_at column found=True