File size: 1,595 Bytes
8f766ee | 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 | import importlib
from fastapi import FastAPI, HTTPException
from registry import TOOLS, BY_NAME
app = FastAPI(title="XTOOLS", description="90two tool server")
_registry_cache = {}
def _load(tool_name: str):
if tool_name not in _registry_cache:
mod = importlib.import_module(f"tools.{tool_name}")
_registry_cache[tool_name] = mod.run
return _registry_cache[tool_name]
def _validate(tool_name: str, params: dict) -> dict:
spec = BY_NAME[tool_name]
cleaned = {}
for p in spec["params"]:
name = p["name"]
if name in params:
cleaned[name] = params[name]
elif p["required"]:
raise HTTPException(status_code=422, detail=f"Missing required param: {name}")
else:
cleaned[name] = p["default"]
return cleaned
@app.get("/")
def health():
return {"status": "ok", "tool_count": len(TOOLS)}
@app.get("/registry")
def get_registry():
"""Full tool manifest -- name, category, auth type, and param schema for every tool."""
return {"tools": TOOLS}
@app.post("/tools/{tool_name}")
def call_tool(tool_name: str, params: dict = None):
if tool_name not in BY_NAME:
raise HTTPException(status_code=404, detail=f"Unknown tool: {tool_name}")
params = params or {}
cleaned = _validate(tool_name, params)
run_fn = _load(tool_name)
try:
return run_fn(cleaned)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail=f"{type(e).__name__}: {e}")
|