Wen Fastmcp
Fastmcp service
Ask AI about Wen Fastmcp
Powered by Claude ยท Grounded in docs
I know everything about Wen Fastmcp. Ask me about installation, configuration, usage, or troubleshooting.
0/500
Reviews
Documentation
FastAPI MCP Demo
A demonstration of MCP (Model Context Protocol) server with FastAPI integration, supporting both STDIO and SSE transports.
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:
- Client establishes SSE connection first (GET
/mcp/sse), getssession_idfrom SSE event - Subsequent POST requests include
session_idas query parameter or header - 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 numbersmultiply(a, b)- Multiply two numbersget_weather(city)- Get weather for a cityreverse_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 improvementssummarize_text(text, max_length)- Summarize texttranslate_text(text, target_language)- Translate textgenerate_questions(topic, num_questions)- Generate practice questionscreate_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())
