| from __future__ import annotations |
|
|
| import os |
| import sys |
| from pathlib import Path |
| from typing import Any |
|
|
| from fastapi import FastAPI |
|
|
|
|
| BASE_DIR = Path(__file__).resolve().parent |
| PLUGIN_DIR = BASE_DIR / "qlib" / "mcp_output" / "mcp_plugin" |
| if str(PLUGIN_DIR) not in sys.path: |
| sys.path.insert(0, str(PLUGIN_DIR)) |
|
|
| app = FastAPI(title="qlib MCP Info App", version="1.0.0") |
|
|
|
|
| def _extract_tools(mcp_instance: Any) -> list[dict[str, str]]: |
| tools_obj = getattr(mcp_instance, "tools", None) |
| if tools_obj is None: |
| return [] |
|
|
| tool_items: list[Any] |
| if isinstance(tools_obj, dict): |
| tool_items = list(tools_obj.values()) |
| elif isinstance(tools_obj, list): |
| tool_items = tools_obj |
| else: |
| try: |
| tool_items = list(tools_obj) |
| except Exception: |
| return [] |
|
|
| result: list[dict[str, str]] = [] |
| for tool in tool_items: |
| name = getattr(tool, "name", None) |
| if name is None: |
| name = getattr(tool, "__name__", "unknown") |
| description = getattr(tool, "description", "") or "" |
| result.append({"name": str(name), "description": str(description)}) |
| return result |
|
|
|
|
| @app.get("/") |
| def root() -> dict[str, Any]: |
| return { |
| "service": "qlib-mcp-deployment", |
| "mcp_entrypoint": "qlib/mcp_output/start_mcp.py", |
| "transport_env": "MCP_TRANSPORT", |
| "default_port": 7860, |
| "note": "This FastAPI app is supplementary and does not run MCP transport.", |
| } |
|
|
|
|
| @app.get("/health") |
| def health() -> dict[str, str]: |
| return {"status": "healthy"} |
|
|
|
|
| @app.get("/tools") |
| def tools() -> dict[str, Any]: |
| from mcp_service import create_app |
|
|
| mcp_instance = create_app() |
| return {"tools": _extract_tools(mcp_instance)} |
|
|
|
|
| if __name__ == "__main__": |
| import uvicorn |
|
|
| port = int(os.getenv("PORT", "7860")) |
| uvicorn.run(app, host="0.0.0.0", port=port) |
|
|