---
source: tests/golden_python.rs
expression: contents
---
"""Multi Scheme API MCP server — generated by mcpify. Do not hand-edit.

Wraps the official `mcp` Python SDK's `FastMCP` server, exposing exactly 3
tools — `search`, `get`, `call` — backed by `mcp_store.db`. Tool bodies
call into `..tools.search_tool`/`..tools.get_tool`/`..tools.call_tool`
(Story P6) and `..auth.auth_manager` (Story P4); this file only wires them
together and adds the logging/timing every tool call gets regardless of
which operation it is. A raised exception inside a tool function is
converted into an MCP error tool result by `FastMCP` itself — no manual
`CallToolResult` construction needed, unlike lower-level SDKs.
"""

from __future__ import annotations

import json
import time
from typing import Any

from mcp.server.fastmcp import Context, FastMCP

from ..auth.auth_manager import AuthManager, header_location_for
from ..auth.request_credentials import RequestCredentials
from ..data.store import cached_store_connection, get_endpoint
from ..http.auth_extractor import ExtractError, extract_request_credentials
from ..tools.call_tool import call_operation
from ..tools.get_tool import get_operation
from ..tools.search_tool import search_operations
from . import logger as _logger_module
from .config import Config
from .errors import NotFoundError

log = _logger_module.get_logger(__name__)

INSTRUCTIONS = (
    "Exposes exactly 3 tools -- search, get, call -- backed by an embedded "
    "semantic database, so you never need the full API surface in context."
)


def build_server(api_version: str, config: Config, auth_manager: AuthManager) -> FastMCP:
    # `streamable_http_path="/"`: this server's own HTTP layer
    # (`http.server.build_http_app`) already mounts the SDK's streamable-HTTP
    # sub-app at `/mcp` — without this, the SDK's own default internal route
    # (also `/mcp`) would double up into `/mcp/mcp`.
    mcp = FastMCP("Multi Scheme API", instructions=INSTRUCTIONS, streamable_http_path="/")

    @mcp.tool(
        description="Semantic search for Multi Scheme API operations using a natural-language query."
    )
    async def search(query: str, limit: int = 5) -> str:
        return await _run_tool(
            "search", api_version, lambda conn: search_operations(conn, query, limit)
        )

    @mcp.tool(
        description="Return the schema, path, method, and documentation for a specific Multi Scheme API operationId."
    )
    async def get(operation_id: str) -> str:
        return await _run_tool(
            "get", api_version, lambda conn: get_operation(conn, operation_id)
        )

    @mcp.tool(
        description="Validate arguments, invoke a live Multi Scheme API API operation, and validate the response."
    )
    async def call(
        operation_id: str, ctx: Context, arguments: dict[str, Any] | None = None
    ) -> str:
        # HTTP transport only: the `mcp` SDK attaches this call's own
        # Starlette `Request` to `ctx.request_context.request` regardless
        # of how long the underlying session's worker lives (it's set
        # fresh per incoming message, not just once at session creation
        # — see `http.server._AuthGateMiddleware`'s doc comment for why
        # this is the one mechanism that actually works here). `None` on
        # stdio, where no such request object is ever attached.
        request_override: RequestCredentials | None = None
        request = ctx.request_context.request
        if request is not None:
            header_location, header_name = header_location_for(config.auth_method)
            try:
                request_override = extract_request_credentials(
                    request.headers, header_location, header_name
                )
            except ExtractError:
                request_override = None

        async def run(conn: Any) -> Any:
            endpoint = await get_endpoint(conn, operation_id)
            if endpoint is None:
                raise NotFoundError(f"unknown operationId '{operation_id}'")
            return await call_operation(
                endpoint, config, auth_manager, operation_id, arguments or {}, request_override
            )

        return await _run_tool_async("call", api_version, run)

    return mcp


async def _run_tool(name: str, api_version: str, fn: Any) -> str:
    async def run(conn: Any) -> Any:
        return await fn(conn)

    return await _run_tool_async(name, api_version, run)


async def _run_tool_async(name: str, api_version: str, fn: Any) -> str:
    started_at = time.monotonic()
    conn = await cached_store_connection(api_version)
    result = await fn(conn)
    duration_ms = (time.monotonic() - started_at) * 1000
    log.info("tool executed", tool=name, duration_ms=round(duration_ms, 2))
    return json.dumps(result, indent=2)
