Spaces:
Configuration error
Configuration error
File size: 4,472 Bytes
e23172f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """Trifecta-Bro MCP Server (MCP 2.0 / FastMCP-style)."""
from __future__ import annotations
import json
from datetime import date as date_type, timedelta
from pathlib import Path
from typing import Any
from mcp.server.mcpserver import MCPServer
from mcp.types import TextContent
mcp = MCPServer("trifecta-bro")
REPO_ROOT = Path(__file__).resolve().parents[1]
PREDICTIONS_DIR = REPO_ROOT / "data" / "predictions"
PDF_DIR = REPO_ROOT / "data" / "pdfs"
PREDICTIONS_DIR.mkdir(parents=True, exist_ok=True)
PDF_DIR.mkdir(parents=True, exist_ok=True)
def _predictions_path(for_date: str) -> Path:
return PREDICTIONS_DIR / f"predictions-{for_date}.json"
def _load_predictions(for_date: str) -> dict[str, Any]:
p = _predictions_path(for_date)
if not p.exists():
raise FileNotFoundError(f"No predictions for {for_date}")
return json.loads(p.read_text())
def _pdfs() -> list[str]:
if not PDF_DIR.exists():
return []
return sorted([p.name for p in PDF_DIR.glob("*.pdf")])
@mcp.tool(name="trifecta.get_predictions")
def get_predictions(for_date: str | None = None) -> list[TextContent]:
"""Return stored trifecta predictions for a date.
If *for_date* is omitted, tomorrow's date (Australia/Adelaide) is used.
"""
target = for_date or (date_type.today() + timedelta(days=1)).isoformat()
try:
payload = _load_predictions(target)
except FileNotFoundError:
return [TextContent(type="text", text=json.dumps({
"error": f"No predictions for {target}",
"date": target,
"tip": "Run generate_predictions.py or run_daily_analysis() first.",
}, indent=2))]
payload["_meta"] = {"date": target, "server": "trifecta-bro-mcp"}
return [TextContent(type="text", text=json.dumps(payload, indent=2))]
@mcp.tool(name="trifecta.get_race")
def get_race(for_date: str, race_number: int) -> list[TextContent]:
"""Return a single race block from stored predictions."""
target = for_date
try:
payload = _load_predictions(target)
except FileNotFoundError:
return [TextContent(type="text", text=json.dumps({
"error": f"No predictions for {target}"
}, indent=2))]
for race in payload.get("races", []):
if int(race.get("race_number", 0)) == int(race_number):
pred = race.get("prediction", {})
out = {
"date": target,
"race_number": race.get("race_number"),
"race_name": race.get("race_name"),
"track": race.get("track"),
"distance": race.get("distance"),
"condition": race.get("condition"),
"race_class": race.get("race_class"),
"prize_money": race.get("prize_money"),
"number_of_runners": race.get("number_of_runners"),
"prediction": pred,
}
return [TextContent(type="text", text=json.dumps(out, indent=2))]
return [TextContent(type="text", text=json.dumps({
"error": f"Race {race_number} not found for {target}"
}, indent=2))]
@mcp.tool(name="trifecta.list_pdfs")
def list_pdfs() -> list[TextContent]:
"""List generated PDF report filenames."""
return [TextContent(type="text", text=json.dumps({"pdfs": _pdfs()}, indent=2))]
@mcp.tool(name="trifecta.get_latest_meetings")
def get_latest_meetings(for_date: str | None = None) -> list[TextContent]:
"""List available meeting dates with stored predictions."""
target = for_date or (date_type.today() + timedelta(days=1)).isoformat()
dates = sorted([p.stem.replace("predictions-", "") for p in PREDICTIONS_DIR.glob("predictions-*.json")])
return [TextContent(type="text", text=json.dumps({
"date": target,
"available_dates": dates,
"count": len(dates),
}, indent=2))]
@mcp.tool(name="trifecta.health")
def health() -> list[TextContent]:
"""Server health and repo status."""
return [TextContent(type="text", text=json.dumps({
"server": "trifecta-bro-mcp",
"status": "ok",
"repo_root": str(REPO_ROOT),
"predictions_dir": str(PREDICTIONS_DIR),
"pdf_dir": str(PDF_DIR),
"predictions_count": len(list(PREDICTIONS_DIR.glob("predictions-*.json"))) if PREDICTIONS_DIR.exists() else 0,
"pdf_count": len(_pdfs()),
}, indent=2))]
if __name__ == "__main__":
import asyncio
asyncio.run(mcp.run_stdio_async())
|