Fast API Auto MCP — native MCP server support via mcp_url

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

fastapi/mcp/__init__.py

@@ -0,0 +1,3 @@
+from fastapi.mcp.server import MCPApp
+
+__all__ = ["MCPApp"]

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

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

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",