MCP Server Creation

class FastApiMCP:
    def __init__(
        self,
        fastapi: FastAPI,
        name: Optional[str] = None,
        description: Optional[str] = None,
        describe_all_responses: bool = False,
        describe_full_response_schema: bool = False,
        http_client: Optional[httpx.AsyncClient] = None,
        include_operations: Optional[List[str]] = None,
        exclude_operations: Optional[List[str]] = None,
        include_tags: Optional[List[str]] = None,
        exclude_tags: Optional[List[str]] = None,
        auth_config: Optional[AuthConfig] = None,
        headers: List[str] = ["authorization"]
    ):
        """
        Create an MCP server from a FastAPI application.

        Parameters:
        - fastapi: The FastAPI application to create an MCP server from
        - name: Name for the MCP server (defaults to app.title)
        - description: Description for the MCP server (defaults to app.description)
        - describe_all_responses: Whether to include all possible response schemas in tool descriptions
        - describe_full_response_schema: Whether to include full JSON schema for responses
        - http_client: Optional custom httpx.AsyncClient for API calls
        - include_operations: List of operation IDs to include as MCP tools (mutually exclusive with exclude_operations)
        - exclude_operations: List of operation IDs to exclude from MCP tools (mutually exclusive with include_operations)
        - include_tags: List of tags to include as MCP tools (mutually exclusive with exclude_tags)
        - exclude_tags: List of tags to exclude from MCP tools (mutually exclusive with include_tags)
        - auth_config: Configuration for MCP authentication
        - headers: HTTP header names to forward from MCP requests (defaults to ['authorization'])
        """

from fastapi import FastAPI, HTTPException, Depends
from fastapi_mcp import FastApiMCP, AuthConfig

app = FastAPI(title="User Management API")

# Create MCP server with operation filtering
mcp_server = FastApiMCP(
    fastapi=app,
    name="User Management MCP",
    include_tags=["users", "auth"],  # Only include endpoints tagged with 'users' or 'auth'
    describe_all_responses=True,     # Include all response schemas
    headers=["authorization", "x-api-key"]  # Forward these headers
)

def mount_http(
    self,
    router: Optional[FastAPI | APIRouter] = None,
    mount_path: str = "/mcp"
) -> None:
    """
    Mount the MCP server with HTTP transport to any FastAPI app or APIRouter.

    Parameters:
    - router: The FastAPI app or APIRouter to mount to (defaults to the original FastAPI app)
    - mount_path: Path where the MCP server will be mounted (defaults to '/mcp')
    """

from fastapi import APIRouter

# Mount to the same app
mcp_server.mount_http()

# Mount to a different app or router
api_router = APIRouter(prefix="/api/v1")
mcp_server.mount_http(router=api_router, mount_path="/mcp-tools")

# The MCP server will be available at the mount path
# AI clients can connect via HTTP to interact with your API

def mount_sse(
    self,
    router: Optional[FastAPI | APIRouter] = None,
    mount_path: str = "/sse"
) -> None:
    """
    Mount the MCP server with SSE transport to any FastAPI app or APIRouter.

    Parameters:
    - router: The FastAPI app or APIRouter to mount to (defaults to the original FastAPI app)
    - mount_path: Path where the MCP server will be mounted (defaults to '/sse')
    """

# Mount SSE transport
mcp_server.mount_sse(mount_path="/mcp-sse")

# Creates endpoints:
# GET /mcp-sse - SSE connection endpoint
# POST /mcp-sse/messages/ - Message posting endpoint

def mount(
    self,
    router: Optional[FastAPI | APIRouter] = None,
    mount_path: str = "/mcp",
    transport: Literal["sse"] = "sse"
) -> None:
    """
    [DEPRECATED] Mount the MCP server to any FastAPI app or APIRouter.
    
    Use mount_http() for HTTP transport or mount_sse() for SSE transport instead.
    """

@property
def server(self) -> Server:
    """The underlying MCP server instance."""

@property
def tools(self) -> List[types.Tool]:
    """List of generated MCP tools from FastAPI endpoints."""

@property
def operation_map(self) -> Dict[str, Dict[str, Any]]:
    """Mapping from tool names to operation details."""

@property
def fastapi(self) -> FastAPI:
    """The FastAPI application instance."""

@property
def name(self) -> str:
    """MCP server name."""

@property
def description(self) -> str:
    """MCP server description."""

# Include only specific operations
mcp_server = FastApiMCP(
    fastapi=app,
    include_operations=["get_user", "create_user", "update_user"]
)

# Exclude specific operations
mcp_server = FastApiMCP(
    fastapi=app,
    exclude_operations=["delete_user", "admin_only_endpoint"]
)

# Include by tags
mcp_server = FastApiMCP(
    fastapi=app,
    include_tags=["public", "user-management"]
)

# Exclude by tags
mcp_server = FastApiMCP(
    fastapi=app,
    exclude_tags=["internal", "admin"]
)

import httpx

# Custom client with authentication
custom_client = httpx.AsyncClient(
    headers={"Authorization": "Bearer token"},
    timeout=30.0
)

mcp_server = FastApiMCP(
    fastapi=app,
    http_client=custom_client
)

mcp_server = FastApiMCP(
    fastapi=app,
    describe_all_responses=True,        # Include all possible HTTP response codes
    describe_full_response_schema=True  # Include complete JSON schema for responses
)

mcp_server = FastApiMCP(
    fastapi=app,
    headers=["authorization", "x-api-key", "x-correlation-id"]
)