mcp_server/server.py

"""FastMCP server exposing the deterministic engine over the MCP protocol.

The agent layer (M5+) talks to these tools as a *client* even though it could import the
engine directly — the client/server boundary is the showcase (golden rule #3). The engine is
built once at import (warm); BioLORD is not loaded unless an unknown pathway appears.

Run standalone (streamable-http on 127.0.0.1):
    python -m mcp_server.server
"""
from __future__ import annotations

from fastmcp import FastMCP

import config
from mcp_server.engine import Engine
from mcp_server.search import PathwaySearch

_engine = Engine()  # warm: loads assets once, reused across calls
_search = PathwaySearch(embed_query=_engine.embed_query)  # optional; disabled without PINECONE_API_KEY


def build_server() -> FastMCP:
    """Construct the FastMCP server. Used by app.py, the demo, and tests."""
    mcp = FastMCP(
        name="semblance-engine",
        instructions="Deterministic GSEA signature comparison. Tools never call an LLM.",
    )

    @mcp.tool()
    def compare_signatures(results: list[dict], params: dict | None = None) -> dict:
        """Compare N canonical GSEA results under cutoff `params`.

        Returns a ComparisonResult: 2N signatures, cosine similarity matrix,
        dendrogram order + linkage + flat clusters, the NES-correlation baseline,
        per-pair drill-downs, and warnings.
        """
        return _engine.compare(results, params)

    @mcp.tool()
    def describe_pathways(names: list[str]) -> dict:
        """Map pathway names to short descriptions (for the interpreter agent)."""
        return _engine.describe(names)

    @mcp.tool()
    def search_pathways(query: str, k: int = 10) -> dict:
        """Semantic free-text search over the MSigDB corpus (optional Pinecone module).

        Returns nearest pathways across collections, or `enabled: False` if the search module
        is not configured. Never used by the core comparison.
        """
        return _search.search(query, k)

    @mcp.tool()
    def match_pathway(name: str, k: int = 10) -> dict:
        """Map a pathway/term name to its nearest known pathways across collections (cross-ontology).

        Embeds the name's concept and returns the closest corpus entries (self excluded), or
        `enabled: False` if the search module is not configured.
        """
        return _search.match(name, k)

    @mcp.tool()
    def health() -> dict:
        """Liveness + asset status (used to poll readiness after subprocess launch)."""
        return {**_engine.health(), "search_enabled": _search.enabled}

    return mcp


mcp = build_server()

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host=config.MCP_HOST, port=config.MCP_PORT)