server.py•5.22 kB
#!/usr/bin/env python3
import sys
import json
import logging
from typing import Dict, List, Optional, Union, Any
import pandas as pd
import os
import tempfile
from jsonrpcserver import Success, Error, method, serve, dispatch
# Import scraper functionality
from scraper import (
fetch_all_transactions,
get_all_nfl_teams,
get_available_transaction_types,
save_to_csv,
save_to_json,
logger
)
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr
)
# Create a handler that logs to stdout for JSON-RPC protocol messages
protocol_logger = logging.getLogger("protocol")
stdout_handler = logging.StreamHandler(sys.stdout)
protocol_logger.addHandler(stdout_handler)
protocol_logger.setLevel(logging.DEBUG)
# List of available tools
@method
def listTools(**kwargs) -> Dict[str, Any]:
"""List available tools in this MCP"""
protocol_logger.info("Listing available tools")
return {
"tools": [
{
"name": "fetch_transactions",
"description": "Fetches NFL transactions based on specified filters"
},
{
"name": "list_teams",
"description": "Lists all NFL teams available for filtering"
},
{
"name": "list_transaction_types",
"description": "Lists all transaction types available for filtering"
}
]
}
# Fetch transactions tool
@method
def fetch_transactions(
start_date: str,
end_date: str,
transaction_type: str = "All",
team: Optional[str] = None,
player: Optional[str] = None,
output_format: str = "json"
) -> Dict[str, Any]:
"""
Fetch NFL transactions based on specified filters
Args:
start_date: Start date in YYYY-MM-DD format
end_date: End date in YYYY-MM-DD format
transaction_type: Type of transaction to filter
team: Team name (optional)
player: Player name (optional)
output_format: Output format (csv, json, or dataframe)
Returns:
Dictionary with transactions data or file path
"""
try:
protocol_logger.info(f"Fetching transactions: {start_date} to {end_date}, type={transaction_type}")
# Fetch transactions
df = fetch_all_transactions(
team=team,
player=player,
start_date=start_date,
end_date=end_date,
transaction_type=transaction_type
)
if df.empty:
return {"status": "success", "message": "No transactions found", "data": []}
# Process based on output format
if output_format == "dataframe":
# For dataframe, return records as JSON
return {"status": "success", "data": json.loads(df.to_json(orient="records"))}
elif output_format == "csv":
# For CSV, save to temp file and return path
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".csv")
file_path = temp_file.name
temp_file.close()
save_to_csv(df, file_path)
return {"status": "success", "file_path": file_path, "count": len(df)}
else: # json
# For JSON, return data directly
return {"status": "success", "data": json.loads(df.to_json(orient="records"))}
except Exception as e:
protocol_logger.error(f"Error in fetch_transactions: {str(e)}")
return Error(message=str(e))
# List teams tool
@method
def list_teams(**kwargs) -> Dict[str, Any]:
"""List all NFL teams available for filtering"""
try:
teams = get_all_nfl_teams()
return {"status": "success", "teams": teams}
except Exception as e:
protocol_logger.error(f"Error in list_teams: {str(e)}")
return Error(message=str(e))
# List transaction types tool
@method
def list_transaction_types(**kwargs) -> Dict[str, Any]:
"""List all transaction types available for filtering"""
try:
transaction_types = get_available_transaction_types()
return {"status": "success", "transaction_types": transaction_types}
except Exception as e:
protocol_logger.error(f"Error in list_transaction_types: {str(e)}")
return Error(message=str(e))
def process_stdin():
"""Process JSON-RPC requests from stdin"""
for line in sys.stdin:
try:
protocol_logger.debug(f"Request: {line.strip()}")
# Dispatch the request
response = dispatch(line)
if response:
protocol_logger.debug(f"Response: {response}")
print(response, flush=True)
except Exception as e:
protocol_logger.error(f"Error processing request: {e}")
error_response = json.dumps(
{"jsonrpc": "2.0", "error": {"code": -32603, "message": str(e)}, "id": None}
)
print(error_response, flush=True)
if __name__ == "__main__":
protocol_logger.info("NFL Transactions MCP server starting")
process_stdin()