FastAPI MCP Demo

A demonstration of MCP (Model Context Protocol) server with FastAPI integration, supporting both STDIO and SSE transports.

English | 简体中文

Installation

# Create virtual environment
uv venv .venv

# Activate virtual environment (Windows)
.venv\Scripts\activate

# Install dependencies
uv pip install -e .

Usage

STDIO Transport (Default)

Run the MCP server using STDIO transport:

# Without authentication
python main.py --transport stdio

# With authentication (single token)
python main.py --transport stdio --token "your-secret-token"

# With authentication (multiple tokens)
python main.py --transport stdio --token "token1" --token "token2"

Client Usage (Python)

Streamable-HTTP Mode (Recommended):

from fastmcp import Client

async def main():
    # Connect to Streamable-HTTP endpoint
    async with Client("http://localhost:8003/mcp", auth="your-token") as client:
        tools = await client.list_tools()
        print(f"Tools: {[t.name for t in tools]}")

        result = await client.call_tool("add", {"a": 5, "b": 3})
        print(f"Result: {result.data}")

asyncio.run(main())

SSE Mode:

from mcp.client.sse import sse_client
from mcp import ClientSession
import asyncio

async def main():
    # SSE endpoint is mounted at /mcp/sse
    async with sse_client(url="http://localhost:8003/mcp/sse", headers={"Authorization": "Bearer your-token"}) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            print(f"Tools: {[t.name for t in tools.tools]}")

            result = await session.call_tool("add", {"a": 5, "b": 3})
            print(f"Result: {result.content[0].text}")

asyncio.run(main())

For a complete example, see examples/sse_client_example.py.

Claude Desktop Configuration

Add to your Claude Desktop config:

{
  "mcpServers": {
    "fastapi-mcp": {
      "command": "python",
      "args": [
        "D:/path/to/fastapi_mcp2/main.py",
        "--transport",
        "stdio",
        "--token",
        "your-secret-token"
      ]
    }
  }
}

SSE/HTTP Transport

Run the MCP server using SSE/HTTP transport via FastAPI:

# Using SSE transport
python main.py --transport sse --host 0.0.0.0 --port 8003

# Using Streamable-HTTP transport
python main.py --transport http --host 0.0.0.0 --port 8003

API Endpoints

All MCP endpoints are mounted under /mcp:

Endpoint	Method	Description
/mcp/sse	GET	SSE stream endpoint
/mcp/messages	POST	JSON-RPC message endpoint

Note: SSE transport requires a complete session flow:

1. Client establishes SSE connection first (GET /mcp/sse), gets session_id from SSE event
2. Subsequent POST requests include session_id as query parameter or header
3. Responses are returned via SSE stream

Authentication

Authentication is enabled by default. Tokens can be configured in multiple ways:

Command Line

# Single token
python main.py --transport stdio --token "your-secret-token"
python main.py --transport sse --token "your-secret-token"

# Multiple tokens
python main.py --transport stdio --token "token1" --token "token2" --token "token3"

Environment Variables

# Single token
set MCP_AUTH_TOKEN=your-secret-token

# Multiple tokens (comma-separated)
set MCP_AUTH_TOKENS=token1,token2,token3

Disable Authentication

python main.py --transport stdio --no-auth
python main.py --transport sse --no-auth

Note: When authentication is enabled but no token is provided, the server will exit immediately with an error message.

Config File

Create a config.json file:

{
  "enabled": true,
  "tokens": ["your-secret-token"],
  "require_auth": true

Load it programmatically:

from src.auth import AuthConfig
config = AuthConfig.from_file("config.json")

Available Tools

• add(a, b) - Add two numbers
• multiply(a, b) - Multiply two numbers
• get_weather(city) - Get weather for a city
• reverse_text(text) - Reverse input text

Available Resources

• config://app-info - Application information

Available Prompts

MCP Prompt component for reusable prompt templates:

• analyze_code(code, language) - Analyze code and provide improvements
• summarize_text(text, max_length) - Summarize text
• translate_text(text, target_language) - Translate text
• generate_questions(topic, num_questions) - Generate practice questions
• create_essay_outline(topic, essay_type) - Create essay outline

Using Prompt Example

import asyncio
from src.server import mcp

async def use_prompt():
    # Get all prompts
    prompts = await mcp.get_prompts()
    print(f"Available prompts: {list(prompts.keys())}")

    # Render prompt template
    detail = prompts.get("analyze_code")
    if detail:
        result = await detail.render(arguments={
            "code": "print('hello')",
            "language": "python"
        })
        print(f"Rendered prompt: {result[0].content.text}")

asyncio.run(use_prompt())

See examples/prompt_example.py for complete example.

API Endpoints

When running with SSE/HTTP transport via FastAPI:

Endpoint	Method	Description
/	GET	Root endpoint with available routes
/health	GET	Health check
/mcp/sse	GET	SSE stream endpoint
/mcp/messages	POST	JSON-RPC message endpoint

Project Structure

src/
├── server.py        # MCP server instance
├── app.py           # FastAPI application
├── auth.py          # Authentication config
├── tools/           # Tool implementations
│   └── __init__.py
├── resources/       # Resource implementations
│   └── __init__.py
└── prompts/         # Prompt implementations
    └── __init__.py
tests/
├── conftest.py      # Pytest fixtures
├── test_auth.py     # Auth tests
└── test_server.py   # Server tests
examples/
├── stdio_client_example.py   # STDIO client example
├── sse_client_example.py     # SSE client example
└── prompt_example.py         # Prompt usage example

src/ ├── server.py # MCP server instance ├── app.py # FastAPI application ├── auth.py # Authentication config ├── tools/ # Tool implementations │ └── init.py └── resources/ # Resource implementations └── init.py tests/ ├── conftest.py # Pytest fixtures ├── test_auth.py # Auth tests └── test_server.py # Server tests examples/ ├── stdio_client_example.py # STDIO client example └── sse_client_example.py # SSE client example

## Adding New Tools

Edit `src/tools/__init__.py`:

```python
from ..server import mcp

@mcp.tool
def my_tool(param: str) -> str:
    """My new tool."""
    return param.upper()

Running Tests

# Run all tests
python -m pytest tests/ -v

# Run specific test
python -m pytest tests/test_auth.py::TestAuthConfig::test_default_config -v

Testing with MCP Client

import asyncio
from fastmcp import Client
from src.server import mcp

async def test():
    async with Client(mcp) as client:
        tools = await client.list_tools()
        result = await client.call_tool("add", {"a": 5, "b": 3})
        print(f"5 + 3 = {result.data}")

if __name__ == "__main__":
    asyncio.run(test())

References

• FastMCP Documentation
• FastAPI Integration