From 632e913171a1d068a36bdb17b666b055274aea11 Mon Sep 17 00:00:00 2001 From: "swe-brain[bot]" Date: Mon, 20 Apr 2026 19:29:26 +0000 Subject: [PATCH] feat: add native MCP server support via mcp_url parameter Adds first-class MCP (Model Context Protocol) support to FastAPI. When mcp_url is set, an MCP server is automatically mounted that exposes all schema-included routes as callable MCP tools. - fastapi/mcp/generator.py: converts APIRoute objects to MCP Tool definitions - fastapi/mcp/server.py: stateless ASGI MCPApp using StreamableHTTP transport - fastapi/applications.py: mcp_url parameter added to FastAPI.__init__ + setup() - pyproject.toml: optional [mcp] dependency group added (mcp>=1.9) - tests/test_mcp.py: 7 integration tests covering all plan requirements --- fastapi/applications.py | 32 ++++++ fastapi/mcp/__init__.py | 3 + fastapi/mcp/generator.py | 93 +++++++++++++++ fastapi/mcp/server.py | 136 ++++++++++++++++++++++ pyproject.toml | 4 + tests/test_mcp.py | 238 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 506 insertions(+) create mode 100644 fastapi/mcp/__init__.py create mode 100644 fastapi/mcp/generator.py create mode 100644 fastapi/mcp/server.py create mode 100644 tests/test_mcp.py diff --git a/fastapi/applications.py b/fastapi/applications.py index 4af1146b0d8f9..0c7ab8837dfa7 100644 --- a/fastapi/applications.py +++ b/fastapi/applications.py @@ -859,6 +859,28 @@ class Item(BaseModel): """ ), ] = True, + mcp_url: Annotated[ + str | None, + Doc( + """ + The path to mount the MCP (Model Context Protocol) server. + + When set, FastAPI automatically generates an MCP server exposing + all routes (with `include_in_schema=True`) as callable tools. + Set to `None` (the default) to disable MCP support. + + Requires the `mcp` extra: `pip install fastapi[mcp]`. + + **Example** + + ```python + from fastapi import FastAPI + + app = FastAPI(mcp_url="/mcp") + ``` + """ + ), + ] = None, **extra: Annotated[ Any, Doc( @@ -882,6 +904,7 @@ class Item(BaseModel): self.root_path_in_servers = root_path_in_servers self.docs_url = docs_url self.redoc_url = redoc_url + self.mcp_url = mcp_url self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url self.swagger_ui_init_oauth = swagger_ui_init_oauth self.swagger_ui_parameters = swagger_ui_parameters @@ -1153,6 +1176,15 @@ async def redoc_html(req: Request) -> HTMLResponse: self.add_route(self.redoc_url, redoc_html, include_in_schema=False) + if self.mcp_url: + try: + from fastapi.mcp.server import MCPApp + except ImportError as exc: + raise ImportError( + "Install fastapi[mcp] to use MCP support: pip install fastapi[mcp]" + ) from exc + self.mount(self.mcp_url, MCPApp(fastapi_app=self)) + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: if self.root_path: scope["root_path"] = self.root_path diff --git a/fastapi/mcp/__init__.py b/fastapi/mcp/__init__.py new file mode 100644 index 0000000000000..3a65db7893b22 --- /dev/null +++ b/fastapi/mcp/__init__.py @@ -0,0 +1,3 @@ +from fastapi.mcp.server import MCPApp + +__all__ = ["MCPApp"] diff --git a/fastapi/mcp/generator.py b/fastapi/mcp/generator.py new file mode 100644 index 0000000000000..767260134ecec --- /dev/null +++ b/fastapi/mcp/generator.py @@ -0,0 +1,93 @@ +from typing import Any + +from mcp.types import Tool +from starlette.routing import BaseRoute + +from fastapi.routing import APIRoute + + +def _resolve_ref(ref: str, schema: dict[str, Any]) -> dict[str, Any]: + """Resolve a $ref within the OpenAPI schema.""" + parts = ref.lstrip("#/").split("/") + node: Any = schema + for part in parts: + node = node[part] + return dict(node) + + +def _build_input_schema( + operation: dict[str, Any], + full_schema: dict[str, Any], +) -> dict[str, Any]: + """Build a JSON Schema input object for an MCP tool from an OpenAPI operation.""" + properties: dict[str, Any] = {} + required: list[str] = [] + + for param in operation.get("parameters", []): + if param.get("in") not in ("path", "query"): + continue + name = param["name"] + param_schema = param.get("schema", {"type": "string"}) + if "$ref" in param_schema: + param_schema = _resolve_ref(param_schema["$ref"], full_schema) + prop: dict[str, Any] = dict(param_schema) + if param.get("description"): + prop["description"] = param["description"] + properties[name] = prop + if param.get("required") or param.get("in") == "path": + required.append(name) + + request_body = operation.get("requestBody") + if request_body: + content = request_body.get("content", {}) + json_content = content.get("application/json", {}) + body_schema = json_content.get("schema", {}) + if "$ref" in body_schema: + body_schema = _resolve_ref(body_schema["$ref"], full_schema) + properties["body"] = body_schema + if request_body.get("required"): + required.append("body") + + result: dict[str, Any] = {"type": "object", "properties": properties} + if required: + result["required"] = required + return result + + +def get_mcp_tools( + routes: list[BaseRoute], + openapi_schema: dict[str, Any], +) -> list[Tool]: + """Convert FastAPI APIRoute objects to MCP Tool definitions.""" + tools: list[Tool] = [] + paths = openapi_schema.get("paths", {}) + + for route in routes: + if not isinstance(route, APIRoute): + continue + if not route.include_in_schema: + continue + + path_item = paths.get(route.path, {}) + for method in route.methods or []: + method_lower = method.lower() + operation = path_item.get(method_lower) + if operation is None: + continue + + description = ( + operation.get("summary") + or operation.get("description") + or f"{method} {route.path}" + ) + input_schema = _build_input_schema(operation, openapi_schema) + + tools.append( + Tool( + name=route.unique_id, + description=description, + inputSchema=input_schema, + ) + ) + + return tools diff --git a/fastapi/mcp/server.py b/fastapi/mcp/server.py new file mode 100644 index 0000000000000..fcf5828dccd52 --- /dev/null +++ b/fastapi/mcp/server.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import json +import logging +import re +import warnings +from typing import TYPE_CHECKING, Any + +import anyio +import httpx +from mcp.server import Server +from mcp.server.streamable_http import StreamableHTTPServerTransport +from mcp.types import CallToolResult, TextContent, Tool +from starlette.routing import BaseRoute +from starlette.types import Receive, Scope, Send + +from fastapi.mcp.generator import get_mcp_tools +from fastapi.routing import APIRoute + +if TYPE_CHECKING: + pass + +logger = logging.getLogger(__name__) + + +class MCPApp: + """ASGI app that exposes FastAPI routes as MCP tools via Streamable HTTP.""" + + def __init__(self, fastapi_app: Any) -> None: + self._fastapi_app = fastapi_app + self._server: Server[None, Any] = _build_server(fastapi_app) + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] == "lifespan": + return + + transport = StreamableHTTPServerTransport( + mcp_session_id=None, + is_json_response_enabled=True, + event_store=None, + ) + + server = self._server + + async def _run_server( + *, task_status: Any = anyio.TASK_STATUS_IGNORED + ) -> None: + async with transport.connect() as (read_stream, write_stream): + task_status.started() + try: + await server.run( + read_stream, + write_stream, + server.create_initialization_options(), + stateless=True, + ) + except Exception: + logger.exception("MCP server session error") + + async with anyio.create_task_group() as tg: + await tg.start(_run_server) + await transport.handle_request(scope, receive, send) + await transport.terminate() + + +def _build_server(fastapi_app: Any) -> Server[None, Any]: + server: Server[None, Any] = Server("FastAPI MCP") + + @server.list_tools() + async def list_tools() -> list[Tool]: + try: + schema = fastapi_app.openapi() + except Exception: + return [] + return get_mcp_tools(fastapi_app.routes, schema) + + @server.call_tool() + async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: + route = _find_route(fastapi_app.routes, name) + if route is None: + return [TextContent(type="text", text=f"Tool not found: {name}")] + + method = next(iter(route.methods or ["GET"])) + path, query_params, body = _extract_call_parts(route, arguments) + + try: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=fastapi_app), + base_url="http://testserver", + ) as client: + resp = await client.request( + method=method, + url=path, + params=query_params or None, + json=body, + ) + return [TextContent(type="text", text=resp.text)] + except Exception as exc: + return [TextContent(type="text", text=f"Error calling {name}: {exc}")] + + return server + + +def _find_route(routes: list[BaseRoute], unique_id: str) -> APIRoute | None: + for route in routes: + if isinstance(route, APIRoute) and route.unique_id == unique_id: + return route + return None + + +def _extract_call_parts( + route: APIRoute, + arguments: dict[str, Any], +) -> tuple[str, dict[str, Any], Any]: + """Return (rendered_path, query_params, body) from tool arguments.""" + path = route.path + query_params: dict[str, Any] = {} + body: Any = None + + path_param_names = set(re.findall(r"\{(\w+)\}", path)) + query_param_names: set[str] = set() + + for dep in route.dependant.path_params: + path_param_names.add(dep.name) + for dep in route.dependant.query_params: + query_param_names.add(dep.name) + + for key, value in arguments.items(): + if key == "body": + body = value + elif key in path_param_names: + path = path.replace(f"{{{key}}}", str(value)) + else: + query_params[key] = value + + return path, query_params, body diff --git a/pyproject.toml b/pyproject.toml index 8d8c4978d8d0c..15a0e92a4af6b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,6 +57,10 @@ Issues = "https://github.com/fastapi/fastapi/issues" Changelog = "https://fastapi.tiangolo.com/release-notes/" [project.optional-dependencies] +mcp = [ + "mcp>=1.9", +] + standard = [ "fastapi-cli[standard] >=0.0.8", "fastar >= 0.9.0", diff --git a/tests/test_mcp.py b/tests/test_mcp.py new file mode 100644 index 0000000000000..35d3889b00e53 --- /dev/null +++ b/tests/test_mcp.py @@ -0,0 +1,238 @@ +"""Tests for FastAPI native MCP support (fastapi/mcp/).""" +import json +from typing import Any + +import httpx +import pytest +from fastapi import FastAPI +from httpx import ASGITransport + +MCP_INIT_PARAMS = { + "protocolVersion": "2024-11-05", + "capabilities": {}, + "clientInfo": {"name": "test-client", "version": "0.1"}, +} + + +def make_client(app: Any) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + follow_redirects=True, + ) + + +async def _mcp_post( + client: httpx.AsyncClient, url: str, body: dict[str, Any] +) -> httpx.Response: + return await client.post( + url, + json=body, + headers={ + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + }, + ) + + +async def _list_tools( + client: httpx.AsyncClient, mcp_url: str +) -> list[dict[str, Any]]: + """Initialize MCP session and return tool list.""" + init_resp = await _mcp_post( + client, + mcp_url, + {"jsonrpc": "2.0", "id": 0, "method": "initialize", "params": MCP_INIT_PARAMS}, + ) + assert init_resp.status_code == 200, f"Init failed: {init_resp.text}" + assert "result" in init_resp.json() + + tools_resp = await _mcp_post( + client, + mcp_url, + {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}, + ) + assert tools_resp.status_code == 200, f"tools/list failed: {tools_resp.text}" + return tools_resp.json().get("result", {}).get("tools", []) + + +@pytest.mark.anyio +async def test_mcp_disabled_by_default() -> None: + app = FastAPI() + + @app.get("/items") + async def list_items() -> list[str]: + return ["a"] + + async with make_client(app) as client: + resp = await client.post("/mcp") + assert resp.status_code == 404 + + +@pytest.mark.anyio +async def test_mcp_custom_url() -> None: + app = FastAPI(mcp_url="/api/mcp") + + @app.get("/items") + async def list_items() -> list[str]: + return ["a"] + + async with make_client(app) as client: + resp = await _mcp_post( + client, + "/api/mcp", + { + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": MCP_INIT_PARAMS, + }, + ) + assert resp.status_code == 200 + assert "result" in resp.json() + + +@pytest.mark.anyio +async def test_mcp_tools_list() -> None: + app = FastAPI(mcp_url="/mcp") + + @app.get("/items", summary="List items") + async def list_items() -> list[str]: + return ["a"] + + @app.post("/items", summary="Create item") + async def create_item(name: str) -> dict[str, str]: + return {"name": name} + + async with make_client(app) as client: + tools = await _list_tools(client, "/mcp") + + tool_names = [t["name"] for t in tools] + assert any("list_items" in n for n in tool_names) + assert any("create_item" in n for n in tool_names) + + +@pytest.mark.anyio +async def test_include_in_schema_false_excluded() -> None: + app = FastAPI(mcp_url="/mcp") + + @app.get("/public") + async def public_route() -> str: + return "ok" + + @app.get("/private", include_in_schema=False) + async def private_route() -> str: + return "secret" + + async with make_client(app) as client: + tools = await _list_tools(client, "/mcp") + + tool_names = [t["name"] for t in tools] + assert any("public" in n for n in tool_names) + assert not any("private" in n for n in tool_names) + + +@pytest.mark.anyio +async def test_mcp_tool_call_get() -> None: + app = FastAPI(mcp_url="/mcp") + + @app.get("/hello", summary="Say hello") + async def hello() -> dict[str, str]: + return {"message": "hello"} + + async with make_client(app) as client: + tools = await _list_tools(client, "/mcp") + hello_tool = next((t for t in tools if "hello" in t["name"]), None) + assert hello_tool is not None + + call_resp = await _mcp_post( + client, + "/mcp", + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": {"name": hello_tool["name"], "arguments": {}}, + }, + ) + assert call_resp.status_code == 200 + result = call_resp.json() + assert "result" in result + content = result["result"]["content"] + assert len(content) > 0 + text = content[0]["text"] + assert "hello" in text + + +@pytest.mark.anyio +async def test_mcp_tool_call_post_with_body() -> None: + from pydantic import BaseModel + + app = FastAPI(mcp_url="/mcp") + + class Item(BaseModel): + name: str + price: float + + @app.post("/items", summary="Create item") + async def create_item(item: Item) -> Item: + return item + + async with make_client(app) as client: + tools = await _list_tools(client, "/mcp") + create_tool = next((t for t in tools if "create_item" in t["name"]), None) + assert create_tool is not None + + call_resp = await _mcp_post( + client, + "/mcp", + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": create_tool["name"], + "arguments": {"body": {"name": "Widget", "price": 9.99}}, + }, + }, + ) + assert call_resp.status_code == 200 + result = call_resp.json() + assert "result" in result + text = result["result"]["content"][0]["text"] + data = json.loads(text) + assert data["name"] == "Widget" + assert data["price"] == pytest.approx(9.99) + + +@pytest.mark.anyio +async def test_mcp_path_params() -> None: + app = FastAPI(mcp_url="/mcp") + + @app.get("/items/{item_id}", summary="Get item") + async def get_item(item_id: int) -> dict[str, int]: + return {"item_id": item_id} + + async with make_client(app) as client: + tools = await _list_tools(client, "/mcp") + get_tool = next((t for t in tools if "get_item" in t["name"]), None) + assert get_tool is not None + + call_resp = await _mcp_post( + client, + "/mcp", + { + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": get_tool["name"], + "arguments": {"item_id": 42}, + }, + }, + ) + assert call_resp.status_code == 200 + result = call_resp.json() + text = result["result"]["content"][0]["text"] + data = json.loads(text) + assert data["item_id"] == 42