from typing import Any, Dict, List
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from db import add_new_product, find_products_by_name, get_all_products, init_db
from tools import safe_eval
app = FastAPI(title="Product MCP Server")
@app.on_event("startup")
def startup():
init_db()
@app.get("/health")
def health() -> Dict[str, str]:
return {"status": "ok"}
class ToolCallRequest(BaseModel):
name: str = Field(..., description="Tool name")
arguments: Dict[str, Any] = Field(default_factory=dict, description="Tool args")
# Метаданные инструментов для LLM
@app.get("/tools")
def list_tools():
return [
{
"name": "list_products",
"description": "Возвращает список доступных товаров",
"parameters": {"type": "object", "properties": {}}
},
{
"name": "find_product",
"description": "Поиск товара по названию",
"parameters": {
"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"]
}
},
{
"name": "add_product",
"description": "Добавить новый товар в базу",
"parameters": {
"type": "object",
"properties": {
"name": {"type": "string"},
"category": {"type": "string"},
"price": {"type": "number"}
},
"required": ["name", "category", "price"]
}
},
{
"name": "calculate",
"description": "Безопасный математический калькулятор",
"parameters": {
"type": "object",
"properties": {"expression": {"type": "string"}},
"required": ["expression"]
}
}
]
@app.post("/call")
def call_tool(req: ToolCallRequest) -> Dict[str, Any]:
try:
if req.name == "list_products":
return {"result": get_all_products()}
if req.name == "find_product":
return {"result": find_products_by_name(req.arguments.get("name", ""))}
if req.name == "add_product":
add_new_product(
req.arguments["name"],
req.arguments["category"],
float(req.arguments["price"]),
)
return {"result": {"ok": True}}
if req.name == "calculate":
return {"result": safe_eval(req.arguments.get("expression", ""))}
raise HTTPException(status_code=404, detail="Tool not found")
except KeyError as e:
raise HTTPException(status_code=400, detail=f"Missing field: {e}") from e
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
