Think Different: Existing API Services to MCP Conversion

MCP server from research gate

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.AsyncClient for 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

MCP Server Architecture

from mcp import Server, Tool, Resource

from mcp.types import TextContent, ImageContent

class CustomMCPServer(Server):

def __init__(self):

super().__init__("custom-server", "1.0.0")

self.register_handlers()

def register_handlers(self):

# Register tools and resources

self.add_tool(self.search_tool)

self.add_resource(self.database_resource)

async def search_tool(self, query: str) -> str:

# Implement search functionality

results = await self.search_service.search(query)

return json.dumps(results)

async def database_resource(self, table: str) -> dict:

# Implement database access

data = await self.db.query(table)

return {"data": data, "schema": self.db.get_schema(table)}

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