fastmcp

1from __future__ import annotations2 3import os4from typing import Any5 6import httpx7from fastmcp import FastMCP8 9 10mcp = FastMCP("__SERVER_NAME__")11 12API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com")13API_TOKEN = os.getenv("API_TOKEN")14REQUEST_TIMEOUT = float(os.getenv("API_TIMEOUT_SECONDS", "20"))15 16 17def _headers() -> dict[str, str]:18    headers = {"Accept": "application/json"}19    if API_TOKEN:20        headers["Authorization"] = f"Bearer {API_TOKEN}"21    return headers22 23 24def _request(method: str, path: str, *, params: dict[str, Any] | None = None) -> Any:25    url = f"{API_BASE_URL.rstrip('/')}/{path.lstrip('/')}"26    with httpx.Client(timeout=REQUEST_TIMEOUT, headers=_headers()) as client:27        response = client.request(method, url, params=params)28        response.raise_for_status()29        return response.json()30 31 32@mcp.tool33def health_check() -> dict[str, Any]:34    """Check whether the upstream API is reachable."""35    payload = _request("GET", "/health")36    return {"base_url": API_BASE_URL, "result": payload}37 38 39@mcp.tool40def get_resource(resource_id: str) -> dict[str, Any]:41    """Fetch one resource by ID from the upstream API."""42    payload = _request("GET", f"/resources/{resource_id}")43    return {"resource_id": resource_id, "data": payload}44 45 46@mcp.tool47def search_resources(query: str, limit: int = 10) -> dict[str, Any]:48    """Search upstream resources by query string."""49    payload = _request("GET", "/resources", params={"q": query, "limit": limit})50    return {"query": query, "limit": limit, "results": payload}51 52 53if __name__ == "__main__":54    mcp.run()55