mcp_server.py•13.3 kB
#!/usr/bin/env python3
"""
MCP Server that wraps the FastAPI sample app
"""
import asyncio
import json
from typing import Any, Dict, List, Optional
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
CallToolRequest,
CallToolResult,
ListToolsRequest,
ListToolsResult,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
)
import httpx
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# FastAPI app base URL
FASTAPI_BASE_URL = "http://localhost:8000"
class FastAPIMCPServer:
def __init__(self):
self.server = Server("fastapi-sample-mcp")
self.client = httpx.AsyncClient()
self.setup_handlers()
def setup_handlers(self):
"""Setup MCP server handlers"""
@self.server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""List available tools"""
return [
Tool(
name="get_app_info",
description="Get basic information about the FastAPI app",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_health",
description="Check the health status of the FastAPI app",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="get_users",
description="Get all users from the FastAPI app",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="create_user",
description="Create a new user",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "User's name"
},
"email": {
"type": "string",
"description": "User's email"
},
"age": {
"type": "integer",
"description": "User's age"
}
},
"required": ["name", "email", "age"]
}
),
Tool(
name="get_user",
description="Get a specific user by ID",
inputSchema={
"type": "object",
"properties": {
"user_id": {
"type": "integer",
"description": "User ID"
}
},
"required": ["user_id"]
}
),
Tool(
name="get_tasks",
description="Get all tasks from the FastAPI app",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="create_task",
description="Create a new task",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Task title"
},
"description": {
"type": "string",
"description": "Task description"
},
"user_id": {
"type": "integer",
"description": "ID of the user who owns this task"
},
"completed": {
"type": "boolean",
"description": "Whether the task is completed",
"default": False
}
},
"required": ["title", "description", "user_id"]
}
),
Tool(
name="get_task",
description="Get a specific task by ID",
inputSchema={
"type": "object",
"properties": {
"task_id": {
"type": "integer",
"description": "Task ID"
}
},
"required": ["task_id"]
}
),
Tool(
name="update_task",
description="Update a task",
inputSchema={
"type": "object",
"properties": {
"task_id": {
"type": "integer",
"description": "Task ID to update"
},
"title": {
"type": "string",
"description": "New task title"
},
"description": {
"type": "string",
"description": "New task description"
},
"user_id": {
"type": "integer",
"description": "ID of the user who owns this task"
},
"completed": {
"type": "boolean",
"description": "Whether the task is completed"
}
},
"required": ["task_id", "title", "description", "user_id", "completed"]
}
),
Tool(
name="delete_task",
description="Delete a task",
inputSchema={
"type": "object",
"properties": {
"task_id": {
"type": "integer",
"description": "Task ID to delete"
}
},
"required": ["task_id"]
}
),
Tool(
name="get_stats",
description="Get statistics about users and tasks",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
]
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls"""
try:
if name == "get_app_info":
response = await self.client.get(f"{FASTAPI_BASE_URL}/")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_health":
response = await self.client.get(f"{FASTAPI_BASE_URL}/health")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_users":
response = await self.client.get(f"{FASTAPI_BASE_URL}/users")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "create_user":
user_data = {
"name": arguments["name"],
"email": arguments["email"],
"age": arguments["age"]
}
response = await self.client.post(f"{FASTAPI_BASE_URL}/users", json=user_data)
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_user":
user_id = arguments["user_id"]
response = await self.client.get(f"{FASTAPI_BASE_URL}/users/{user_id}")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_tasks":
response = await self.client.get(f"{FASTAPI_BASE_URL}/tasks")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "create_task":
task_data = {
"title": arguments["title"],
"description": arguments["description"],
"user_id": arguments["user_id"],
"completed": arguments.get("completed", False)
}
response = await self.client.post(f"{FASTAPI_BASE_URL}/tasks", json=task_data)
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_task":
task_id = arguments["task_id"]
response = await self.client.get(f"{FASTAPI_BASE_URL}/tasks/{task_id}")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "update_task":
task_id = arguments["task_id"]
task_data = {
"title": arguments["title"],
"description": arguments["description"],
"user_id": arguments["user_id"],
"completed": arguments["completed"]
}
response = await self.client.put(f"{FASTAPI_BASE_URL}/tasks/{task_id}", json=task_data)
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "delete_task":
task_id = arguments["task_id"]
response = await self.client.delete(f"{FASTAPI_BASE_URL}/tasks/{task_id}")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_stats":
response = await self.client.get(f"{FASTAPI_BASE_URL}/stats")
result = response.json()
return [TextContent(type="text", text=json.dumps(result, indent=2))]
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except httpx.HTTPStatusError as e:
error_msg = f"HTTP error {e.response.status_code}: {e.response.text}"
return [TextContent(type="text", text=error_msg)]
except Exception as e:
error_msg = f"Error calling tool {name}: {str(e)}"
logger.error(error_msg)
return [TextContent(type="text", text=error_msg)]
async def main():
"""Main entry point"""
mcp_server = FastAPIMCPServer()
async with stdio_server() as (read_stream, write_stream):
await mcp_server.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="fastapi-sample-mcp",
server_version="1.0.0",
capabilities=mcp_server.server.get_capabilities(
notification_options=None,
experimental_capabilities=None,
),
),
)
if __name__ == "__main__":
asyncio.run(main())