from fastapi import FastAPI, Request, Response, Depends, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, validator
from typing import Optional, Dict, Any
import os
import json
import time
app = FastAPI(title="BMI MCP Server", version="1.0.0")
# Optional auth: set API_KEY in env and pass header x-api-key
API_KEY = os.getenv("API_KEY")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ---------- MCP models ----------
class ToolSchema(BaseModel):
name: str
description: str
inputSchema: Dict[str, Any]
class MCPHello(BaseModel):
type: str = "hello"
protocol: str = "mcp-1.0"
tools: list[ToolSchema]
class InvokeRequest(BaseModel):
tool: str = Field(..., description="Tool name, e.g. 'bmiCalculator'")
params: Dict[str, Any] = Field(..., description="Input params for the tool")
class BMIInput(BaseModel):
weight: float = Field(..., gt=0, description="kg")
height: float = Field(..., gt=0, description="height value in m or cm")
unit: Optional[str] = Field("m", description="height unit: 'm' or 'cm'")
@validator("height")
def height_reasonable(cls, v):
if v <= 0:
raise ValueError("height must be > 0")
return v
def require_api_key(request: Request):
"""If API_KEY is set, require header x-api-key to match."""
if API_KEY:
key = request.headers.get("x-api-key")
if key != API_KEY:
raise HTTPException(status_code=401, detail="Invalid or missing API key.")
# ---------- tool impl ----------
def calc_bmi(weight: float, height: float, unit: str = "m") -> Dict[str, Any]:
if unit.lower() == "cm":
height = height / 100.0
if height <= 0 or weight <= 0:
raise ValueError("Weight and height must be positive.")
bmi = weight / (height * height)
bmi = round(bmi, 2)
if bmi < 18.5:
category = "Underweight"
elif bmi < 24.0:
category = "Normal"
elif bmi < 28.0:
category = "Overweight (Pre-obese)"
else:
category = "Obese"
advice_map = {
"Underweight": "建议营养评估与适度力量训练。",
"Normal": "保持良好饮食与运动习惯。",
"Overweight (Pre-obese)": "建议控制总热量并增加有氧运动。",
"Obese": "建议个体化体重管理与医学评估。"
}
return {
"bmi": bmi,
"category": category,
"advice": advice_map[category]
}
# ---------- SSE: /mcp ----------
@app.get("/mcp")
def mcp_stream(request: Request, _: None = Depends(require_api_key)):
"""
MCP discovery endpoint via SSE. Emits a hello + tools JSON and keeps the
connection alive with pings.
"""
def event_generator():
hello = MCPHello(
tools=[
ToolSchema(
name="bmiCalculator",
description="Calculate BMI from weight (kg) and height (m or cm).",
inputSchema={
"type": "object",
"required": ["weight", "height"],
"properties": {
"weight": {"type": "number", "description": "kg"},
"height": {"type": "number", "description": "m or cm"},
"unit": {"type": "string", "enum": ["m", "cm"], "default": "m"}
}
}
)
]
)
payload = json.dumps(hello.dict(), ensure_ascii=False)
yield f"event: mcp\ndata: {payload}\n\n"
# keep-alive pings
import time
while True:
client = request.client
if client is None:
break
yield "event: ping\ndata: {}\n\n"
time.sleep(20)
return StreamingResponse(event_generator(), media_type="text/event-stream")
# ---------- tool execution: /invoke ----------
@app.post("/invoke")
def invoke_tool(req: InvokeRequest, _: None = Depends(require_api_key)):
if req.tool != "bmiCalculator":
raise HTTPException(status_code=404, detail=f"Unknown tool: {req.tool}")
try:
inp = BMIInput(**req.params)
result = calc_bmi(inp.weight, inp.height, inp.unit)
return {"ok": True, "tool": req.tool, "result": result}
except Exception as e:
return JSONResponse(status_code=400, content={"ok": False, "error": str(e)})
# ---------- health ----------
@app.get("/healthz")
def healthz():
return {"status": "ok", "version": "1.0.0"}
