An MCP server, or Model Context Protocol server, acts as a bridge, enabling AI models to interact with external tools and data sources through a standardized interface. It simplifies the process of connecting AI applications to various services and resources, making it easier for developers to build and deploy AI-powered applications.
We all have tons of API services and servers already. However now due to AgenticAI and multiple Agent applications with MCP. It is getting difficult if not totally impossible to connect the same API services. So I thought, how can we convert our existing API services into MCP capable.
This guide provides a complete end-to-end solution for converting FastAPI services (both sync and async) into MCP (Model Context Protocol) servers that can be called by LLM agents.
Overview
What is MCP? Model Context Protocol (MCP) is a standardized protocol that enables LLM applications to securely connect to data sources and tools. It provides a way for AI assistants to interact with external services through tools, resources, and prompts.
Why Convert FastAPI to MCP?
- Standardized interface for LLM agents
- Better security and access control
- Structured tool definitions
- Resource management capabilities
- Protocol-level error handling
Architecture
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ LLM Agent │────│ MCP Server │────│ FastAPI Service│
│ Application │ │ (Bridge) │ │ (Your APIs) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Step-by-Step Implementation
1. Project Structure
fastapi-mcp-conversion/
├── fastapi_services.py # Your existing FastAPI services
├── mcp_server.py # MCP server implementation
├── requirements.txt # Dependencies
├── pyproject.toml # Project configuration
├── setup.sh # Setup script
├── run_fastapi.sh # FastAPI runner
├── run_mcp_server.sh # MCP server runner
├── test_integration.sh # Integration tests
├── test_mcp_client.py # Test client
├── production_deployment.py # Production setup
└── README.md # This documentation
2. FastAPI Services (Source)
Your existing FastAPI services should include:
- Sync endpoints: Traditional REST endpoints
- Async endpoints: Asynchronous operations
- Proper error handling: HTTP status codes and error messages
- Pydantic models: Request/response validation
3. MCP Server Implementation
The MCP server acts as a bridge between your FastAPI services and LLM agents:
Key Components:
- Tools: Represent callable functions (your API endpoints)
- Resources: Represent data sources or documentation
- Server: Handles MCP protocol communication
Tool Definition Example:
Tool(
name="get_users",
description="Get all users from the system",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
4. Conversion Process
Step 1: Analyze FastAPI Endpoints
- Identify all endpoints (GET, POST, PUT, DELETE)
- Extract parameters and request/response models
- Categorize sync vs async operations
- Document endpoint purposes and behaviors
Step 2: Create MCP Tools
For each FastAPI endpoint, create an MCP tool:
- Map endpoint parameters to tool input schema
- Define clear descriptions
- Handle both sync and async operations
- Implement error handling and validation
Step 3: Implement HTTP Client
- Use
httpx.AsyncClientfor making requests to FastAPI - Handle timeouts and retries
- Manage connection pooling
- Convert HTTP errors to MCP-compatible responses
Step 4: Resource Management
Define resources for:
- API documentation
- Data schemas
- Service status
- Configuration information
5. Configuration and Deployment
Environment Variables
FASTAPI_BASE_URL=http://localhost:8000
MCP_SERVER_NAME=fastapi-mcp-server
LOG_LEVEL=INFO
Claude Desktop Configuration
Add to your Claude Desktop config:
{
"mcpServers": {
"fastapi-mcp-server": {
"command": "python",
"args": ["mcp_server.py"],
"env": {
"FASTAPI_BASE_URL": "http://localhost:8000"
}
}
}
}
6. Running the System
Development Setup
# 1. Setup environment
./setup.sh
# 2. Start FastAPI service
./run_fastapi.sh
# 3. Start MCP server (in another terminal)
./run_mcp_server.sh
# 4. Test integration
./test_integration.sh
Production Deployment
# Using Docker Compose
docker-compose up -d
# Or manual deployment
python production_deployment.py
7. Testing and Validation
Unit Tests
Test individual components:
- FastAPI endpoints
- MCP tool implementations
- HTTP client functionality
- Error handling
Integration Tests
Test end-to-end functionality:
- MCP server startup
- Tool execution
- Resource access
- Error scenarios
Load Testing
Test performance:
- Concurrent requests
- Memory usage
- Response times
- Connection limits
8. LLM Agent Integration
Example Agent Usage
# In your LLM agent application
async def process_user_request(user_input: str):
if "create user" in user_input.lower():
# Extract user data from natural language
user_data = extract_user_data(user_input)
# Call MCP tool
result = await mcp_session.call_tool("create_user", user_data)
# Process result and respond
return format_response(result)
9. Benefits of This Approach
For Developers
- Reuse existing APIs: No need to rewrite FastAPI services
- Standardized interface: Consistent way to expose APIs to LLMs
- Better error handling: Protocol-level error management
- Documentation: Auto-generated tool descriptions
For LLM Agents
- Type safety: Structured input/output schemas
- Discovery: Automatic tool and resource discovery
- Reliability: Robust error handling and retries
- Security: Controlled access to backend services
10. Advanced Features
Authentication and Authorization
# Add authentication to MCP server
async def authenticate_request(request_data):
# Implement your auth logic
pass
Rate Limiting
# Implement rate limiting
from asyncio import Semaphore
class RateLimitedMCPServer:
def __init__(self, max_concurrent=10):
self.semaphore = Semaphore(max_concurrent)
Caching
# Add caching for frequently accessed data
from functools import lru_cache
@lru_cache(maxsize=100)
async def cached_get_users():
# Implementation
pass
Monitoring and Metrics
# Track usage metrics
class MCPServerMetrics:
def record_request(self, tool_name, response_time, success):
# Implementation
pass
11. Troubleshooting
Common Issues
- Connection refused: FastAPI service not running
- Tool not found: Incorrect tool name mapping
- Schema validation: Mismatched input parameters
- Timeout errors: Network or processing delays
Debug Mode
# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
Health Checks
# Implement health check endpoint
@app.get("/health")
async def health_check():
return {"status": "healthy", "timestamp": datetime.utcnow()}
12. Performance Optimization
Connection Pooling
# Configure HTTP client for optimal performance
client = httpx.AsyncClient(
timeout=30.0,
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
Async Processing
# Use asyncio for concurrent operations
async def process_multiple_requests(requests):
tasks = [process_request(req) for req in requests]
return await asyncio.gather(*tasks)
13. Security Considerations
Input Validation
- Validate all input parameters
- Sanitize user data
- Implement parameter bounds checking
Access Control
- Implement authentication for sensitive operations
- Use API keys or tokens
- Log all access attempts
Error Information
- Don't expose internal system details in errors
- Use generic error messages for security
- Log detailed errors internally
14. Scaling and Production
Horizontal Scaling
- Deploy multiple MCP server instances
- Use load balancer for FastAPI services
- Implement health checks
Monitoring
- Track request/response metrics
- Monitor error rates
- Set up alerting
Backup and Recovery
- Implement graceful shutdown
- Handle service restarts
- Maintain service availability
Full example
2. Protocol Implementation:
Message Types:
- Tool calls: Request tool execution with parameters
- Resource requests: Access to structured data
- Capability discovery: Query available tools and resources
- Authentication: Secure access control
Communication Flow:
- Client discovers server capabilities
- Client authenticates with server
- Client requests tools/resources
- Server processes requests and returns results
- Error handling and retry mechanisms
3. Security Considerations:
- Authentication: OAuth2, API keys, or custom auth
- Authorization: Role-based access control
- Input validation: Sanitize all inputs
- Rate limiting: Prevent abuse and overload
- Audit logging: Track all interactions
Development Best Practices:
Server Implementation:
- Async programming: Use asyncio for concurrent handling
- Error handling: Comprehensive error responses
- Documentation: OpenAPI specs for tool definitions
- Versioning: Support multiple protocol versions
Testing Framework:
import pytest
from mcp.testing import MCPTestClient
@pytest.mark.asyncio
async def test_search_tool():
server = CustomMCPServer()
client = MCPTestClient(server)
response = await client.call_tool("search", {"query": "test"})
assert response.status == "success"
assert "results" in response.data
import pytest
from mcp.testing import MCPTestClient
@pytest.mark.asyncio
async def test_search_tool():
server = CustomMCPServer()
client = MCPTestClient(server)
response = await client.call_tool("search", {"query": "test"})
assert response.status == "success"
assert "results" in response.data
Conclusion
This complete guide provides everything needed to convert FastAPI services into MCP servers that can be used by LLM agents. The approach maintains the benefits of your existing FastAPI implementation while providing a standardized interface for AI applications.
The conversion process is straightforward and allows for gradual migration of services. Start with a few key endpoints and expand coverage as needed.
