# server.py
import os
from typing import Any, Dict
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from fastapi.middleware.cors import CORSMiddleware
from .mgnify_client import MGnifyClient
from .schemas import StudySearchParams, RunId, AnalysesParams
from fastapi import FastAPI, Request, Response
import json
import uvicorn
load_dotenv()
host = os.getenv("MCP_HOST", "0.0.0.0")
port = int(os.getenv("MCP_PORT", "8173"))
ALLOWED_ORIGINS = os.getenv("MCP_CORS_ORIGINS", "http://localhost:3000").split(",")
# app = FastAPI(title="MGnify MCP Server")
#
#
# app.add_middleware(
# CORSMiddleware,
# allow_origins=[o.strip() for o in ALLOWED_ORIGINS],
# allow_methods=["POST", "OPTIONS"],
# allow_headers=["content-type"],
# )
#
# @app.post("/jsonrpc")
# async def jsonrpc_endpoint(request: Request):
# """
# Accept JSON-RPC 2.0 requests from frontend and forward them to FastMCP.
# """
# payload = await request.json()
# try:
# # handle_jsonrpc is part of FastMCP’s core
# result = await server.handle_jsonrpc(payload)
# return Response(
# content=json.dumps(result),
# media_type="application/json",
# status_code=200,
# )
# except Exception as e:
# return Response(
# content=json.dumps({"error": str(e)}),
# media_type="application/json",
# status_code=500,
# )
#
# def main():
# print(f"🚀 MGnify MCP HTTP server running at http://{host}:{port}/jsonrpc")
# uvicorn.run(app, host=host, port=port)
# # server.run(transport="sse")
#
#
# # ---- Main --------------------------------------------------------------------
# if __name__ == "__main__":
# main()
# def main():
# # server.run()
# # host = os.getenv("MCP_HOST", "0.0.0.0")
# # port = int(os.getenv("MCP_PORT", "8173"))
#
# # Check if FastMCP supports this:
# # server.run(transport="http", host=host, port=port)
# server.run(transport="sse")
#
# # If FastMCP runs its own HTTP server, this will include the app with CORS
# # server.run()
# server = FastMCP("mgnify-mcp")
server = FastMCP("mgnify-mcp",
host=host,
port=port,
)
client = MGnifyClient()
# --- CORS on underlying app ---------------------------------------------------
# app = getattr(server, "app", None) or getattr(server, "asgi_app", None)
# app = getattr(server, "app", None) or getattr(server, "asgi_app", None)
# if app:
# allowed = os.getenv("MCP_CORS_ORIGINS", "http://localhost:3000").split(",")
# app.add_middleware(
# CORSMiddleware,
# allow_origins=["*"],
# # allow_origins=[o.strip() for o in allowed],
# allow_methods=["POST", "OPTIONS"],
# allow_headers=["content-type"],
# )
# ---- Resources ---------------------------------------------------------------
if hasattr(server, "resource"):
@server.resource("mgnify://biomes")
def res_biomes() -> Dict[str, Any]:
return client.biomes()
@server.resource("mgnify://runs/{run_id}")
def res_run(run_id: str) -> Dict[str, Any]:
return client.run(run_id)
@server.resource("mgnify://studies")
def res_studies_default() -> Dict[str, Any]:
return client.search_studies(query=None, biome=None, page=1, size=25)
# ---- Tools -------------------------------------------------------------------
@server.tool()
def search_studies(params: StudySearchParams) -> Dict[str, Any]:
try:
return client.search_studies(params.query, params.biome, params.page, params.size)
except Exception as e:
return {"error": str(e)}
@server.tool()
def get_run(run: RunId) -> Dict[str, Any]:
try:
return client.run(run.id)
except Exception as e:
return {"error": str(e)}
@server.tool()
def list_analyses(params: AnalysesParams) -> Dict[str, Any]:
try:
return client.analyses_for_study(params.study_id, params.type, params.page, params.size)
except Exception as e:
return {"error": str(e)}
# ---- Prompts -----------------------------------------------------------------
@server.prompt()
def explain_api_plan() -> str:
return (
"You are an assistant that translates natural language questions into MGnify API calls. "
"When given a question, return a JSON object with: rationale, endpoints[], and suggested follow-ups."
)
TOOL_REGISTRY = {
"search_studies": lambda args: search_studies(StudySearchParams(**(args or {}))),
"get_run": lambda args: get_run(RunId(**(args or {}))),
"list_analyses": lambda args: list_analyses(AnalysesParams(**(args or {}))),
# add more tools here as you create them
}
app = FastAPI(title="MGnify MCP Server")
app.add_middleware(
CORSMiddleware,
allow_origins=[o.strip() for o in ALLOWED_ORIGINS],
allow_methods=["POST", "OPTIONS"],
allow_headers=["content-type"],
)
def jsonrpc_ok(rpc_id, result):
return {"jsonrpc": "2.0", "id": rpc_id, "result": result}
def jsonrpc_err(rpc_id, code, message, data=None):
err = {"code": code, "message": message}
if data is not None:
err["data"] = data
return {"jsonrpc": "2.0", "id": rpc_id, "error": err}
@app.post("/jsonrpc")
async def jsonrpc_endpoint(request: Request):
payload = await request.json()
rpc_id = payload.get("id")
method = payload.get("method")
params = payload.get("params", {})
try:
if method == "tools/call":
name = params.get("name")
args = params.get("arguments", {})
tool = TOOL_REGISTRY.get(name)
if not tool:
return jsonrpc_err(rpc_id, -32601, f"Unknown tool: {name}")
result = tool(args)
# If your tool already returns {"error": "..."} you can normalize it here if you prefer
return jsonrpc_ok(rpc_id, result)
# (Optional) support a simple resource read pattern if you want it:
# elif method == "resources/read":
# uri = params.get("uri")
# ... (map mgnify:// URIs to client.* calls) ...
return jsonrpc_err(rpc_id, -32601, f"Unknown method: {method}")
except Exception as e:
# Don’t leak stack traces in prod; include a request fingerprint instead
return jsonrpc_err(rpc_id, -32000, "Internal error", {"detail": str(e)})
def main():
print(f"🚀 MGnify MCP HTTP server at http://{host}:{port}/jsonrpc")
uvicorn.run(app, host=host, port=port)
if __name__ == "__main__":
main()
# import os
# from typing import Any, Dict
# from dotenv import load_dotenv
# from fastapi import FastAPI, Request, Response
# from fastapi.middleware.cors import CORSMiddleware
# import uvicorn
# import asyncio
# import json
#
# from mcp.server.fastmcp import FastMCP
# from .mgnify_client import MGnifyClient
# from .schemas import StudySearchParams, RunId, AnalysesParams
#
# load_dotenv()
#
# # ---------------------------------------------------------------------------
# # Config
# # ---------------------------------------------------------------------------
# HOST = os.getenv("MCP_HOST", "0.0.0.0")
# PORT = int(os.getenv("MCP_PORT", "8173"))
# ALLOWED_ORIGINS = os.getenv("MCP_CORS_ORIGINS", "http://localhost:3000").split(",")
#
# # ---------------------------------------------------------------------------
# # Core setup
# # ---------------------------------------------------------------------------
# server = FastMCP("mgnify-mcp")
# client = MGnifyClient()
#
# # ---------------------------------------------------------------------------
# # Define MCP resources & tools
# # ---------------------------------------------------------------------------
# if hasattr(server, "resource"):
# @server.resource("mgnify://biomes")
# def res_biomes() -> Dict[str, Any]:
# return client.biomes()
#
# @server.resource("mgnify://runs/{run_id}")
# def res_run(run_id: str) -> Dict[str, Any]:
# return client.run(run_id)
#
# @server.resource("mgnify://studies")
# def res_studies_default() -> Dict[str, Any]:
# return client.search_studies(None, None, 1, 25)
#
# @server.tool()
# def search_studies(params: StudySearchParams) -> Dict[str, Any]:
# return client.search_studies(params.query, params.biome, params.page, params.size)
#
# @server.tool()
# def get_run(run: RunId) -> Dict[str, Any]:
# return client.run(run.id)
#
# @server.tool()
# def list_analyses(params: AnalysesParams) -> Dict[str, Any]:
# return client.analyses_for_study(params.study_id, params.type, params.page, params.size)
#
# @server.prompt()
# def explain_api_plan() -> str:
# return (
# "You are an assistant that translates natural language questions into MGnify API calls. "
# "When given a question, return a JSON object with: rationale, endpoints[], and suggested follow-ups."
# )
#
# # ---------------------------------------------------------------------------
# # FastAPI wrapper + manual JSON-RPC forwarding
# # ---------------------------------------------------------------------------
# app = FastAPI(title="MGnify MCP Server")
#
# app.add_middleware(
# CORSMiddleware,
# allow_origins=[o.strip() for o in ALLOWED_ORIGINS],
# allow_methods=["POST", "OPTIONS"],
# allow_headers=["content-type"],
# )
#
# @app.post("/jsonrpc")
# async def jsonrpc_endpoint(request: Request):
# """
# Accept JSON-RPC 2.0 requests from frontend and forward them to FastMCP.
# """
# payload = await request.json()
# try:
# # handle_jsonrpc is part of FastMCP’s core
# result = await server.handle_jsonrpc(payload)
# return Response(
# content=json.dumps(result),
# media_type="application/json",
# status_code=200,
# )
# except Exception as e:
# return Response(
# content=json.dumps({"error": str(e)}),
# media_type="application/json",
# status_code=500,
# )
#
# # ---------------------------------------------------------------------------
# # Entry
# # ---------------------------------------------------------------------------
# def main():
# print(f"🚀 MGnify MCP HTTP server running at http://{HOST}:{PORT}/jsonrpc")
# uvicorn.run(app, host=HOST, port=PORT)
#
# if __name__ == "__main__":
# main()
