qlib / app.py
ghh1125's picture
Upload 14 files
f948afb verified
Raw
History Blame Contribute Delete
1.9 kB
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)