Skip to content

3.3. MCP

What problem does MCP solve?

The Model Context Protocol gives tools a process and framework-independent discovery/invocation contract. Instead of importing the Ops Copilot's Python functions, another compatible client can connect to an MCP server and receive the same schemas.

MCP does not automatically make a tool safe. Authentication, authorization, transport security, timeouts, rate limits, schema validation, and tool implementation still matter.

Which tools does the course server expose?

Only six read capabilities cross the MCP boundary:

for tool in (
    tools.list_incidents,
    tools.get_incident,
    tools.get_service_status,
    tools.search_service_logs,
    memory.get_runbook,
    memory.search_runbooks,
):
    mcp.add_tool(tool)

restart_service and resolve_incident remain in-process because their ADK confirmation context and audit identity are part of the write boundary. Separating read and write surfaces makes authorization easier to reason about.

Which transports are supported?

mcp_server.py defaults to stdio for an agent that launches its own child process. It also supports SSE and streamable HTTP with the same bounded graceful-shutdown policy as A2A:

def _run_http(transport: Literal["sse", "streamable-http"]) -> None:
    """Serve MCP HTTP with the same bounded SIGTERM drain as A2A."""
    app = mcp.sse_app() if transport == "sse" else mcp.streamable_http_app()
    uvicorn.run(
        app,
        host=mcp.settings.host,
        port=mcp.settings.port,
        log_level=mcp.settings.log_level.lower(),
        timeout_graceful_shutdown=int(settings.drain_timeout_s),
    )


def main() -> None:
    """Run over stdio by default or bounded Uvicorn for HTTP transports."""
    transport = os.environ.get("MCP_TRANSPORT", "stdio")
    if transport not in {"stdio", "sse", "streamable-http"}:
        raise ValueError(f"Unsupported MCP_TRANSPORT: {transport!r}")
    if transport == "stdio":
        mcp.run("stdio")
        return
    _run_http(cast("Literal['sse', 'streamable-http']", transport))

Use the repository tasks:

cd agents/python
mise run mcp       # stdio
mise run mcp:http  # streamable HTTP on 127.0.0.1:8000/mcp

The Kubernetes MCP deployment uses the same module with MCP_HOST=0.0.0.0, port 8000, and streamable HTTP.

HTTP transport keeps FastMCP's DNS-rebinding protection enabled even when it binds to all interfaces. Secure defaults accept only loopback authorities and the course's agentops-mcp service names; browser origins remain limited to loopback HTTP. MCP_ALLOWED_HOSTS is a comma-separated full override, not an addition, so each deployment can narrow the authorities it expects without falling back to *.

How does the ADK client choose a transport?

def ops_mcp_toolset(url: str | None = None) -> McpToolset:
    """Return the toolset over local stdio or a gateway streamable-HTTP URL.

    Both transports carry the course's explicit deadlines (Chapter 4.5): a hung
    MCP server or gateway then fails a tool call fast instead of hanging a turn.
    """
    endpoint = url or settings.mcp_url
    if endpoint:
        # A secured gateway route (Ch. 5.5) authenticates the caller by bearer
        # token; the default local route needs no header.
        headers = {"Authorization": f"Bearer {settings.mcp_token.get_secret_value()}"} if settings.mcp_token else None
        return McpToolset(
            connection_params=StreamableHTTPConnectionParams(
                url=endpoint,
                headers=headers,
                timeout=settings.tool_timeout_s,
                sse_read_timeout=settings.tool_timeout_s,
            ),
        )
    return McpToolset(
        connection_params=StdioConnectionParams(
            # Use the current interpreter so it works inside the project's virtualenv.
            server_params=StdioServerParameters(command=sys.executable, args=["-m", "agent.mcp_server"]),
            timeout=settings.tool_timeout_s,
        ),
    )

The exact excerpt is build-checked against mcp_client.py. Using sys.executable keeps the child in the same locked Python environment; configured HTTP clients also carry the course deadline and optional bearer token.

When does the root agent use MCP?

Local/offline development registers the six Python read tools directly. When AGENT_MCP_URL is present, the composition root replaces them with one remote McpToolset:

def _read_tools() -> list[ToolUnion]:
    if settings.mcp_url:
        return [ops_mcp_toolset(settings.mcp_url)]
    return [*ALL_TOOLS, *KNOWLEDGE_TOOLS]

In Kubernetes:

AGENT_MCP_URL=http://agentgateway:3000/mcp

On the host with the loopback wrapper, use http://127.0.0.1:3000/mcp. The deployed call path is agent -> agentgateway -> agentops-mcp:8000; guarded writes and instruction-only skills remain local to the agent.

Why put agentgateway between the client and server?

Chapter 5 adds a stable policy point for tool allowlists, fail-closed behavior, rate limits, logs, and traces. The Python tool implementation stays the same; only the configured MCP endpoint changes.

What is the MCP checkpoint?

cd agents/python
uv run pytest tests/test_mcp.py tests/test_tools.py

Verify exactly six tools, all supported transports, rejection of an unknown transport, DNS-rebinding rejection for an untrusted Host, stdio construction, HTTP construction, and conditional root-agent composition. A test that merely opens TCP port 8000 is not an MCP protocol test.