"""Main MCP Server implementation using stdio transport"""
import asyncio
import json
import sys
from typing import Any, Dict, List, Optional
from mcp.server import Server, InitializationOptions
from mcp.types import Tool, TextContent, ServerCapabilities, ToolsCapability
from mcp.server.stdio import stdio_server
from src.openf1_client import OpenF1Client
class OpenF1MCPServer:
"""MCP Server for openF1.org API"""
def __init__(self):
self.server = Server("openf1-mcp")
self.client = None
self._register_tools()
self._register_list_tools()
def _register_list_tools(self):
"""Register the list_tools handler"""
@self.server.list_tools()
async def list_tools():
return self.get_tools()
def _register_tools(self):
"""Register all available tools"""
@self.server.call_tool()
async def handle_tool_call(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls from the client"""
try:
# Always create a fresh client for each request to avoid session timeout issues
async with OpenF1Client() as client:
result = await self._run_tool(name, arguments, client)
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def _run_tool(self, name: str, arguments: Dict[str, Any], client: OpenF1Client) -> Any:
"""Run the actual tool logic"""
if name == "list_drivers":
driver_number = arguments.get("driver_number")
return await client.get_drivers(driver_number)
elif name == "list_teams":
return await client.get_teams()
elif name == "list_races":
year = arguments.get("year")
return await client.get_races(year)
elif name == "list_meetings":
year = arguments.get("year")
return await client.get_meetings(year)
elif name == "list_results":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_results(session_key, driver_number)
elif name == "list_sessions":
session_key = arguments.get("session_key")
meeting_key = arguments.get("meeting_key")
return await client.get_sessions(session_key, meeting_key)
elif name == "list_laps":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_laps(session_key, driver_number)
elif name == "list_stints":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_stints(session_key, driver_number)
elif name == "list_pit_stops":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_pit_stops(session_key, driver_number)
elif name == "get_weather":
session_key = arguments.get("session_key")
return await client.get_weather(session_key)
elif name == "list_incidents":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_incidents(session_key, driver_number)
elif name == "get_car_data":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_car_data(session_key, driver_number)
elif name == "list_positions":
session_key = arguments.get("session_key")
driver_number = arguments.get("driver_number")
return await client.get_positions(session_key, driver_number)
elif name == "get_driver_standings":
session_key = arguments.get("session_key")
return await client.get_driver_standings(session_key)
elif name == "get_constructor_standings":
session_key = arguments.get("session_key")
return await client.get_constructor_standings(session_key)
elif name == "get_server_status":
return {
"status": "operational",
"server_name": "openf1-mcp",
"server_version": "1.0.0",
"tools_count": len(self.get_tools()),
"api_endpoint": "https://api.openf1.org/v1",
"message": "Server is running and ready to process requests"
}
elif name == "verify_client_connection":
from datetime import datetime, timezone
return {
"connected": True,
"timestamp": datetime.now(timezone.utc).isoformat(),
"verification": "Client connection verified",
"ready_for_requests": True
}
elif name == "get_available_tools":
tools_list = []
for tool in self.get_tools():
tools_list.append({
"name": tool.name,
"description": tool.description,
"has_schema": tool.inputSchema is not None
})
return {
"total_tools": len(tools_list),
"tools": tools_list
}
else:
raise ValueError(f"Unknown tool: {name}")
def get_tools(self) -> List[Tool]:
"""Return list of available tools"""
return [
Tool(
name="list_drivers",
description="Fetch F1 drivers. Optionally filter by driver number.",
inputSchema={
"type": "object",
"properties": {
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="list_teams",
description="Fetch F1 teams.",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="list_races",
description="Fetch F1 races. Optionally filter by year.",
inputSchema={
"type": "object",
"properties": {
"year": {
"type": "integer",
"description": "Filter by year (e.g., 2024)"
}
}
}
),
Tool(
name="list_meetings",
description="Fetch F1 race meetings. Optionally filter by year.",
inputSchema={
"type": "object",
"properties": {
"year": {
"type": "integer",
"description": "Filter by year (e.g., 2024)"
}
}
}
),
Tool(
name="list_results",
description="Fetch race results. Optionally filter by session or driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Filter by session key"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="list_sessions",
description="Fetch F1 sessions (practice, qualifying, race). Optionally filter by session_key or meeting_key.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Filter by session key"
},
"meeting_key": {
"type": "integer",
"description": "Filter by meeting key"
}
}
}
),
Tool(
name="list_laps",
description="Fetch lap data from a session. Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key (required for filtering)"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="list_stints",
description="Fetch stint data (tire stints). Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key (required for filtering)"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="list_pit_stops",
description="Fetch pit stop data from a session. Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key (required for filtering)"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="get_weather",
description="Fetch weather data for a session.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key"
}
}
}
),
Tool(
name="list_incidents",
description="Fetch incident data (collisions, penalties, etc.) from a session. Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="get_car_data",
description="Fetch car telemetry data (throttle, brake, DRS, etc.). Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="list_positions",
description="Fetch position data (live positions during session). Optionally filter by driver.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key"
},
"driver_number": {
"type": "integer",
"description": "Filter by driver number"
}
}
}
),
Tool(
name="get_driver_standings",
description="Get driver championship standings. Optionally filter by session_key.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key to get standings after that session"
}
}
}
),
Tool(
name="get_constructor_standings",
description="Get constructor/team championship standings. Optionally filter by session_key.",
inputSchema={
"type": "object",
"properties": {
"session_key": {
"type": "integer",
"description": "Session key to get standings after that session"
}
}
}
),
Tool(
name="get_server_status",
description="Get the current status of the MCP server, including version and availability.",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="verify_client_connection",
description="Verify that the client is properly connected to the MCP server.",
inputSchema={
"type": "object",
"properties": {}
}
),
Tool(
name="get_available_tools",
description="Get a list of all available tools provided by this MCP server.",
inputSchema={
"type": "object",
"properties": {}
}
)
]
async def run(self):
"""Start the MCP server using stdio transport"""
# Create initialization options with tools capability
init_options = InitializationOptions(
server_name="openf1-mcp",
server_version="1.0.0",
capabilities=ServerCapabilities(
tools=ToolsCapability()
)
)
# Run the server
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
init_options
)
async def main():
"""Main entry point"""
server = OpenF1MCPServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main())
