# Directory Structure
```
├── .gitattributes
├── .gitignore
├── LICENSE
├── package-lock.json
├── package.json
├── python
│ └── f1_data.py
├── README.md
├── src
│ └── index.ts
├── test_server.js
└── tsconfig.json
```
# Files
--------------------------------------------------------------------------------
/.gitattributes:
--------------------------------------------------------------------------------
```
# Auto detect text files and perform LF normalization
* text=auto
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Dependencies
node_modules/
package-lock.json
yarn.lock
# Build outputs
dist/
build/
*.js
!jest.config.js
!test_server.js
# TypeScript
*.tsbuildinfo
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
ENV/
# IDE
.idea/
.vscode/*
!.vscode/settings.json
!.vscode/tasks.json
!.vscode/launch.json
!.vscode/extensions.json
# OS
.DS_Store
.AppleDouble
.LSOverride
._*
# Logs
logs
*.log
npm-debug.log*
yarn-debug.log*
yarn-error.log*
# Environment variables
.env
.env.test
.env.local
.env.development.local
.env.test.local
.env.production.local
# Test coverage
coverage/
cache/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Formula One MCP Server
This Model Context Protocol (MCP) server provides access to Formula One data and statistics using the FastF1 Python library. It allows you to access race calendars, event information, session results, driver data, lap times, telemetry, and championship standings through a clean MCP interface.
## Features
- Get Formula One race calendars for specific seasons
- Retrieve detailed information about Grand Prix events
- Get session results (Race, Qualifying, Practice)
- Access driver information and statistics
- Analyze driver performance with lap times and telemetry data
- Compare multiple drivers' performance
- Get championship standings for drivers and constructors
## Prerequisites
- Node.js 18 or later
- Python 3.8 or later
- FastF1 library
## Installation
### 1. Install Python dependencies
```bash
pip install fastf1 pandas numpy
```
### 2. Install Node.js dependencies
```bash
cd f1-mcp-server
npm install
```
### 3. Build the TypeScript code
```bash
npm run build
```
### 4. Add to MCP settings
Add the following to your Cline MCP settings file (`~/Library/Application Support/Code/User/globalStorage/saoudrizwan.claude-dev/settings/cline_mcp_settings.json`):
```json
{
"mcpServers": {
"formula1": {
"command": "node",
"args": ["/Users/rakeshgangwar/Documents/Cline/MCP/f1-mcp-server/build/index.js"],
"disabled": false,
"autoApprove": []
}
}
}
```
## Available Tools
### 1. `get_event_schedule`
Get Formula One race calendar for a specific season.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
### 2. `get_event_info`
Get detailed information about a specific Formula One Grand Prix.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `identifier` (string): Event name or round number (e.g., "Monaco" or "7")
### 3. `get_session_results`
Get results for a specific Formula One session.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `event_identifier` (string): Event name or round number (e.g., "Monaco" or "7")
- `session_name` (string): Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")
### 4. `get_driver_info`
Get information about a specific Formula One driver.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `event_identifier` (string): Event name or round number (e.g., "Monaco" or "7")
- `session_name` (string): Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")
- `driver_identifier` (string): Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")
### 5. `analyze_driver_performance`
Analyze a driver's performance in a Formula One session.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `event_identifier` (string): Event name or round number (e.g., "Monaco" or "7")
- `session_name` (string): Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")
- `driver_identifier` (string): Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")
### 6. `compare_drivers`
Compare performance between multiple Formula One drivers.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `event_identifier` (string): Event name or round number (e.g., "Monaco" or "7")
- `session_name` (string): Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")
- `drivers` (string): Comma-separated list of driver codes (e.g., "HAM,VER,LEC")
### 7. `get_telemetry`
Get telemetry data for a specific Formula One lap.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `event_identifier` (string): Event name or round number (e.g., "Monaco" or "7")
- `session_name` (string): Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")
- `driver_identifier` (string): Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")
- `lap_number` (number, optional): Lap number (gets fastest lap if not provided)
### 8. `get_championship_standings`
Get Formula One championship standings.
**Parameters:**
- `year` (number): Season year (e.g., 2023)
- `round_num` (number, optional): Round number (gets latest standings if not provided)
## Example Usage
Once the server is added to your MCP settings and running, you can use these tools with Cline to access Formula One data.
Example queries:
- "Show me the 2023 Formula One race calendar"
- "Get the results from the 2022 Monaco Grand Prix"
- "Compare Hamilton and Verstappen's performance in the 2021 British Grand Prix"
- "Show me the telemetry data from Leclerc's fastest lap in the 2023 Italian Grand Prix qualifying"
- "What are the current F1 championship standings?"
## Data Source
This server uses the [FastF1](https://github.com/theOehrly/Fast-F1) Python library, which provides access to official Formula 1 timing data, car telemetry, and session results.
## License
MIT
```
--------------------------------------------------------------------------------
/tsconfig.json:
--------------------------------------------------------------------------------
```json
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"esModuleInterop": true,
"outDir": "./build",
"strict": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "**/*.spec.ts"]
}
```
--------------------------------------------------------------------------------
/package.json:
--------------------------------------------------------------------------------
```json
{
"name": "f1-mcp-server",
"version": "1.0.0",
"description": "Formula One MCP server using FastF1",
"main": "build/index.js",
"type": "module",
"scripts": {
"build": "tsc && chmod +x build/index.js",
"start": "node build/index.js",
"dev": "nodemon --exec node --loader ts-node/esm src/index.ts"
},
"author": "",
"license": "MIT",
"dependencies": {
"@modelcontextprotocol/sdk": "^1.7.0"
},
"devDependencies": {
"@types/node": "^20.11.24",
"nodemon": "^3.1.0",
"ts-node": "^10.9.2",
"typescript": "^5.3.3"
}
}
```
--------------------------------------------------------------------------------
/test_server.js:
--------------------------------------------------------------------------------
```javascript
#!/usr/bin/env node
/**
* Test script for the Formula One MCP server
* This script will:
* 1. Import the server index.js file
* 2. Test that it loads correctly
*/
console.log('Testing Formula One MCP Server...');
console.log('This script will import the main server file.');
// We don't actually need to do anything here as the server
// code automatically starts the MCP server when imported
import('./build/index.js')
.then(() => {
console.log('Server imported successfully.');
console.log('The server is now running in the background.');
console.log('You can now use the Formula One tools in Cline.');
console.log('Press Ctrl+C to exit this test script.');
})
.catch((error) => {
console.error('Failed to import server code:', error);
process.exit(1);
});
// Keep the script running
process.stdin.resume();
console.log('Waiting for server messages...');
```
--------------------------------------------------------------------------------
/python/f1_data.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
FastF1 MCP Server - Python Bridge
This script provides functions to access Formula 1 data using the FastF1 library.
It is designed to be called from the Node.js MCP server.
"""
import sys
import json
import traceback
import fastf1
import pandas as pd
import numpy as np
from datetime import datetime
# Configure FastF1 cache
fastf1.Cache.enable_cache('~/Documents/Cline/MCP/f1-mcp-server/cache')
def json_serial(obj):
"""Helper function to convert non-JSON serializable objects to strings"""
if isinstance(obj, (datetime, pd.Timestamp)):
return obj.isoformat()
if isinstance(obj, (np.integer, np.floating)):
return float(obj) if isinstance(obj, np.floating) else int(obj)
if pd.isna(obj):
return None
return str(obj)
def get_event_schedule(year):
"""Get the event schedule for a specified season"""
try:
year = int(year)
schedule = fastf1.get_event_schedule(year)
# Convert DataFrame to JSON serializable format
result = []
for _, row in schedule.iterrows():
event_dict = row.to_dict()
# Clean and convert non-serializable values
clean_dict = {k: json_serial(v) for k, v in event_dict.items()}
result.append(clean_dict)
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def get_event_info(year, identifier):
"""Get information about a specific event"""
try:
year = int(year)
# Identifier can be event name or round number
if identifier.isdigit():
event = fastf1.get_event(year, int(identifier))
else:
event = fastf1.get_event(year, identifier)
# Convert Series to dict and clean non-serializable values
event_dict = event.to_dict()
clean_dict = {k: json_serial(v) for k, v in event_dict.items()}
return {"status": "success", "data": clean_dict}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def get_session_results(year, event_identifier, session_name):
"""Get results for a specific session"""
try:
year = int(year)
session = fastf1.get_session(year, event_identifier, session_name)
session.load(telemetry=False) # Load session without telemetry for faster results
# Get results as a DataFrame
results = session.results
# Convert results to JSON serializable format
result_list = []
for driver_num, result in results.items():
driver_result = result.to_dict()
# Clean and convert non-serializable values
clean_dict = {k: json_serial(v) for k, v in driver_result.items()}
result_list.append(clean_dict)
return {"status": "success", "data": result_list}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def get_driver_info(year, event_identifier, session_name, driver_identifier):
"""Get information about a specific driver"""
try:
year = int(year)
session = fastf1.get_session(year, event_identifier, session_name)
session.load(telemetry=False) # Load session without telemetry for faster results
driver_info = session.get_driver(driver_identifier)
# Convert to JSON serializable format
driver_dict = driver_info.to_dict()
clean_dict = {k: json_serial(v) for k, v in driver_dict.items()}
return {"status": "success", "data": clean_dict}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def analyze_driver_performance(year, event_identifier, session_name, driver_identifier):
"""Analyze a driver's performance in a session"""
try:
year = int(year)
session = fastf1.get_session(year, event_identifier, session_name)
session.load()
# Get laps for the specified driver
driver_laps = session.laps.pick_driver(driver_identifier)
# Basic statistics
fastest_lap = driver_laps.pick_fastest()
# Calculate average lap time (excluding outliers)
valid_lap_times = []
for _, lap in driver_laps.iterrows():
if lap['LapTime'] is not None and not pd.isna(lap['LapTime']):
valid_lap_times.append(lap['LapTime'].total_seconds())
avg_lap_time = sum(valid_lap_times) / len(valid_lap_times) if valid_lap_times else None
# Format lap time as minutes:seconds.milliseconds
formatted_fastest = str(fastest_lap['LapTime']) if not pd.isna(fastest_lap['LapTime']) else None
# Get all lap times
lap_times = []
for _, lap in driver_laps.iterrows():
lap_dict = {
"LapNumber": int(lap['LapNumber']) if not pd.isna(lap['LapNumber']) else None,
"LapTime": str(lap['LapTime']) if not pd.isna(lap['LapTime']) else None,
"Compound": lap['Compound'] if not pd.isna(lap['Compound']) else None,
"TyreLife": int(lap['TyreLife']) if not pd.isna(lap['TyreLife']) else None,
"Stint": int(lap['Stint']) if not pd.isna(lap['Stint']) else None,
"FreshTyre": bool(lap['FreshTyre']) if not pd.isna(lap['FreshTyre']) else None,
"LapStartTime": json_serial(lap['LapStartTime']) if not pd.isna(lap['LapStartTime']) else None
}
lap_times.append(lap_dict)
# Format results
result = {
"DriverCode": fastest_lap['Driver'] if not pd.isna(fastest_lap['Driver']) else None,
"TotalLaps": len(driver_laps),
"FastestLap": formatted_fastest,
"AverageLapTime": avg_lap_time,
"LapTimes": lap_times
}
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def compare_drivers(year, event_identifier, session_name, drivers):
"""Compare performance between multiple drivers"""
try:
year = int(year)
drivers_list = drivers.split(",")
session = fastf1.get_session(year, event_identifier, session_name)
session.load()
driver_comparisons = []
for driver in drivers_list:
# Get laps and fastest lap for each driver
driver_laps = session.laps.pick_driver(driver)
fastest_lap = driver_laps.pick_fastest()
# Calculate average lap time
valid_lap_times = []
for _, lap in driver_laps.iterrows():
if lap['LapTime'] is not None and not pd.isna(lap['LapTime']):
valid_lap_times.append(lap['LapTime'].total_seconds())
avg_lap_time = sum(valid_lap_times) / len(valid_lap_times) if valid_lap_times else None
# Format lap time as string
formatted_fastest = str(fastest_lap['LapTime']) if not pd.isna(fastest_lap['LapTime']) else None
# Compile driver data
driver_data = {
"DriverCode": driver,
"FastestLap": formatted_fastest,
"FastestLapNumber": int(fastest_lap['LapNumber']) if not pd.isna(fastest_lap['LapNumber']) else None,
"TotalLaps": len(driver_laps),
"AverageLapTime": avg_lap_time
}
driver_comparisons.append(driver_data)
return {"status": "success", "data": driver_comparisons}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def get_telemetry(year, event_identifier, session_name, driver_identifier, lap_number=None):
"""Get telemetry data for a specific lap or fastest lap"""
try:
year = int(year)
session = fastf1.get_session(year, event_identifier, session_name)
session.load()
# Get laps for the specified driver
driver_laps = session.laps.pick_driver(driver_identifier)
# Get the specific lap or fastest lap
if lap_number:
lap = driver_laps[driver_laps['LapNumber'] == int(lap_number)].iloc[0]
else:
lap = driver_laps.pick_fastest()
# Get telemetry data
telemetry = lap.get_telemetry()
# Convert to JSON serializable format
telemetry_dict = telemetry.to_dict(orient='records')
clean_data = []
for item in telemetry_dict:
clean_item = {k: json_serial(v) for k, v in item.items()}
clean_data.append(clean_item)
# Add lap information
lap_info = {
"LapNumber": int(lap['LapNumber']) if not pd.isna(lap['LapNumber']) else None,
"LapTime": str(lap['LapTime']) if not pd.isna(lap['LapTime']) else None,
"Compound": lap['Compound'] if not pd.isna(lap['Compound']) else None,
"TyreLife": int(lap['TyreLife']) if not pd.isna(lap['TyreLife']) else None
}
result = {
"lapInfo": lap_info,
"telemetry": clean_data
}
return {"status": "success", "data": result}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def get_championship_standings(year, round_num=None):
"""Get championship standings for drivers and constructors"""
try:
year = int(year)
# Create Ergast API client
ergast = fastf1.ergast.Ergast()
# Get Ergast API data
if round_num:
drivers_standings = ergast.get_driver_standings(season=year, round=round_num).content[0]
constructor_standings = ergast.get_constructor_standings(season=year, round=round_num).content[0]
else:
drivers_standings = ergast.get_driver_standings(season=year).content[0]
constructor_standings = ergast.get_constructor_standings(season=year).content[0]
# Convert driver standings to JSON serializable format
drivers_list = []
for _, row in drivers_standings.iterrows():
driver_dict = row.to_dict()
clean_dict = {k: json_serial(v) for k, v in driver_dict.items()}
drivers_list.append(clean_dict)
# Convert constructor standings to JSON serializable format
constructors_list = []
for _, row in constructor_standings.iterrows():
constructor_dict = row.to_dict()
clean_dict = {k: json_serial(v) for k, v in constructor_dict.items()}
constructors_list.append(clean_dict)
return {
"status": "success",
"data": {
"drivers": drivers_list,
"constructors": constructors_list
}
}
except Exception as e:
return {"status": "error", "message": str(e), "traceback": traceback.format_exc()}
def main():
"""Main function to parse arguments and call appropriate function"""
if len(sys.argv) < 2:
print(json.dumps({"status": "error", "message": "No function specified"}))
return
function_name = sys.argv[1]
args = sys.argv[2:] if len(sys.argv) > 2 else []
functions = {
"get_event_schedule": get_event_schedule,
"get_event_info": get_event_info,
"get_session_results": get_session_results,
"get_driver_info": get_driver_info,
"analyze_driver_performance": analyze_driver_performance,
"compare_drivers": compare_drivers,
"get_telemetry": get_telemetry,
"get_championship_standings": get_championship_standings
}
if function_name in functions:
result = functions[function_name](*args)
print(json.dumps(result))
else:
print(json.dumps({"status": "error", "message": f"Unknown function: {function_name}"}))
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/index.ts:
--------------------------------------------------------------------------------
```typescript
#!/usr/bin/env node
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
CallToolRequestSchema,
ErrorCode,
ListToolsRequestSchema,
McpError,
} from '@modelcontextprotocol/sdk/types.js';
import { spawn } from 'child_process';
import path from 'path';
import { fileURLToPath } from 'url';
// Get the directory name of the current module
const __dirname = path.dirname(fileURLToPath(import.meta.url));
// Path to the Python script
const pythonScriptPath = path.resolve(__dirname, '../python/f1_data.py');
/**
* Execute a Python function from the f1_data.py script
* @param functionName - The function to call in the Python script
* @param args - Arguments to pass to the function
* @returns The result from the Python script
*/
async function executePythonFunction(functionName: string, args: string[] = []): Promise<any> {
return new Promise((resolve, reject) => {
// Use 'python3' for macOS/Linux, 'python' for Windows
const pythonProcess = spawn('python3', [pythonScriptPath, functionName, ...args]);
let result = '';
let error = '';
pythonProcess.stdout.on('data', (data) => {
result += data.toString();
});
pythonProcess.stderr.on('data', (data) => {
error += data.toString();
});
pythonProcess.on('close', (code) => {
if (code !== 0) {
reject(new Error(`Python process exited with code ${code}: ${error}`));
} else {
try {
resolve(JSON.parse(result));
} catch (parseError) {
reject(new Error(`Failed to parse Python output: ${result}\n${parseError}`));
}
}
});
});
}
class FormulaOneServer {
private server: Server;
constructor() {
this.server = new Server(
{
name: 'formula-one-server',
version: '1.0.0',
},
{
capabilities: {
resources: {},
tools: {},
},
}
);
// Set up the tool handlers
this.setupToolHandlers();
// Error handling
this.server.onerror = (error) => console.error('[MCP Error]', error);
process.on('SIGINT', async () => {
await this.server.close();
process.exit(0);
});
}
/**
* Set up the tool handlers for the MCP server
*/
private setupToolHandlers() {
// List available tools
this.server.setRequestHandler(ListToolsRequestSchema, async () => ({
tools: [
{
name: 'get_event_schedule',
description: 'Get Formula One race calendar for a specific season',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
},
required: ['year'],
},
},
{
name: 'get_event_info',
description: 'Get detailed information about a specific Formula One Grand Prix',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
},
required: ['year', 'identifier'],
},
},
{
name: 'get_session_results',
description: 'Get results for a specific Formula One session',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
event_identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
session_name: {
type: 'string',
description: 'Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")',
},
},
required: ['year', 'event_identifier', 'session_name'],
},
},
{
name: 'get_driver_info',
description: 'Get information about a specific Formula One driver',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
event_identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
session_name: {
type: 'string',
description: 'Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")',
},
driver_identifier: {
type: 'string',
description: 'Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")',
},
},
required: ['year', 'event_identifier', 'session_name', 'driver_identifier'],
},
},
{
name: 'analyze_driver_performance',
description: 'Analyze a driver\'s performance in a Formula One session',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
event_identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
session_name: {
type: 'string',
description: 'Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")',
},
driver_identifier: {
type: 'string',
description: 'Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")',
},
},
required: ['year', 'event_identifier', 'session_name', 'driver_identifier'],
},
},
{
name: 'compare_drivers',
description: 'Compare performance between multiple Formula One drivers',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
event_identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
session_name: {
type: 'string',
description: 'Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")',
},
drivers: {
type: 'string',
description: 'Comma-separated list of driver codes (e.g., "HAM,VER,LEC")',
},
},
required: ['year', 'event_identifier', 'session_name', 'drivers'],
},
},
{
name: 'get_telemetry',
description: 'Get telemetry data for a specific Formula One lap',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
event_identifier: {
type: 'string',
description: 'Event name or round number (e.g., "Monaco" or "7")',
},
session_name: {
type: 'string',
description: 'Session name (e.g., "Race", "Qualifying", "Sprint", "FP1", "FP2", "FP3")',
},
driver_identifier: {
type: 'string',
description: 'Driver identifier (number, code, or name; e.g., "44", "HAM", "Hamilton")',
},
lap_number: {
type: 'number',
description: 'Lap number (optional, gets fastest lap if not provided)',
},
},
required: ['year', 'event_identifier', 'session_name', 'driver_identifier'],
},
},
{
name: 'get_championship_standings',
description: 'Get Formula One championship standings',
inputSchema: {
type: 'object',
properties: {
year: {
type: 'number',
description: 'Season year (e.g., 2023)',
},
round_num: {
type: 'number',
description: 'Round number (optional, gets latest standings if not provided)',
},
},
required: ['year'],
},
},
],
}));
// Handle tool calls
this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
// Type assertion for arguments to avoid TypeScript errors
type EventScheduleArgs = { year: number };
type EventInfoArgs = { year: number; identifier: string };
type SessionResultsArgs = { year: number; event_identifier: string; session_name: string };
type DriverInfoArgs = { year: number; event_identifier: string; session_name: string; driver_identifier: string };
type AnalyzeDriverArgs = { year: number; event_identifier: string; session_name: string; driver_identifier: string };
type CompareDriversArgs = { year: number; event_identifier: string; session_name: string; drivers: string };
type TelemetryArgs = { year: number; event_identifier: string; session_name: string; driver_identifier: string; lap_number?: number };
type ChampionshipArgs = { year: number; round_num?: number };
try {
let result;
switch (name) {
case 'get_event_schedule': {
const typedArgs = args as EventScheduleArgs;
result = await executePythonFunction('get_event_schedule', [typedArgs.year.toString()]);
break;
}
case 'get_event_info': {
const typedArgs = args as EventInfoArgs;
result = await executePythonFunction('get_event_info', [
typedArgs.year.toString(),
typedArgs.identifier.toString(),
]);
break;
}
case 'get_session_results': {
const typedArgs = args as SessionResultsArgs;
result = await executePythonFunction('get_session_results', [
typedArgs.year.toString(),
typedArgs.event_identifier.toString(),
typedArgs.session_name.toString(),
]);
break;
}
case 'get_driver_info': {
const typedArgs = args as DriverInfoArgs;
result = await executePythonFunction('get_driver_info', [
typedArgs.year.toString(),
typedArgs.event_identifier.toString(),
typedArgs.session_name.toString(),
typedArgs.driver_identifier.toString(),
]);
break;
}
case 'analyze_driver_performance': {
const typedArgs = args as AnalyzeDriverArgs;
result = await executePythonFunction('analyze_driver_performance', [
typedArgs.year.toString(),
typedArgs.event_identifier.toString(),
typedArgs.session_name.toString(),
typedArgs.driver_identifier.toString(),
]);
break;
}
case 'compare_drivers': {
const typedArgs = args as CompareDriversArgs;
result = await executePythonFunction('compare_drivers', [
typedArgs.year.toString(),
typedArgs.event_identifier.toString(),
typedArgs.session_name.toString(),
typedArgs.drivers.toString(),
]);
break;
}
case 'get_telemetry': {
const typedArgs = args as TelemetryArgs;
const telemetryArgs = [
typedArgs.year.toString(),
typedArgs.event_identifier.toString(),
typedArgs.session_name.toString(),
typedArgs.driver_identifier.toString(),
];
if (typedArgs.lap_number !== undefined) {
telemetryArgs.push(typedArgs.lap_number.toString());
}
result = await executePythonFunction('get_telemetry', telemetryArgs);
break;
}
case 'get_championship_standings': {
const typedArgs = args as ChampionshipArgs;
const standingsArgs = [typedArgs.year.toString()];
if (typedArgs.round_num !== undefined) {
standingsArgs.push(typedArgs.round_num.toString());
}
result = await executePythonFunction('get_championship_standings', standingsArgs);
break;
}
default:
throw new McpError(ErrorCode.MethodNotFound, `Unknown tool: ${name}`);
}
if (result.status === 'error') {
return {
content: [
{
type: 'text',
text: `Error: ${result.message}\n\n${result.traceback || ''}`,
},
],
isError: true,
};
}
return {
content: [
{
type: 'text',
text: JSON.stringify(result.data, null, 2),
},
],
};
} catch (error) {
console.error(`Error executing tool ${name}:`, error);
return {
content: [
{
type: 'text',
text: `Error: ${error instanceof Error ? error.message : String(error)}`,
},
],
isError: true,
};
}
});
}
/**
* Run the MCP server
*/
async run() {
const transport = new StdioServerTransport();
await this.server.connect(transport);
console.error('Formula One MCP server running on stdio');
}
}
const server = new FormulaOneServer();
server.run().catch(console.error);
```