mcp_stdio.py

"""Claude Desktop stdio entry point for PCSWMM Engineering MCP.

This process exposes the same deterministic tool registry used by server.py,
without starting FastAPI or opening a TCP port. Standard output is reserved
exclusively for the MCP protocol.
"""
from __future__ import annotations

import logging
import os
from pathlib import Path
import sys

from mcp.server.fastmcp import FastMCP

import agent as agent_module
from pcswmm_tools import PCSWMM_TOOL_REGISTRY as TOOL_REGISTRY


SERVER_NAME = "pcswmm-engineering"
PROJECT_DIR = Path(__file__).resolve().parent

# Independent SWMM verification uses its own isolated worker interpreter.
worker_python = PROJECT_DIR / ".swmm-worker-venv" / "Scripts" / "python.exe"
if worker_python.is_file():
    os.environ.setdefault("SWMM_WORKER_PYTHON", str(worker_python))

# Never write application logging to stdout: stdout carries JSON-RPC/MCP traffic.
logging.basicConfig(
    level=os.environ.get("PCSWMM_MCP_LOG_LEVEL", "WARNING").upper(),
    stream=sys.stderr,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)

mcp = FastMCP(
    SERVER_NAME,
    instructions=(
        "Local-first deterministic stormwater engineering copilot. "
        "Connect either an active PCSWMM SDK evidence package, an existing "
        "deterministic Calgary evidence folder, or PySWMM/EPA SWMM evidence. "
        "Validate the evidence, perform engineering QA/QC and Calgary screening, "
        "configure report details, and generate engineer-review SWMR deliverables. "
        "Do not invent hydraulic results or missing project criteria."
    ),
)

for tool_name, tool_function in TOOL_REGISTRY.items():
    mcp.tool(name=tool_name)(tool_function)


@mcp.tool()
def agent_analyze(
    question: str,
    provider: str = "local",
    model: str = "",
    session_id: str = "",
    api_key: str = "",
    base_url: str = "",
) -> dict:
    """Run optional narrative orchestration over the deterministic MCP tools."""
    return agent_module.run_agent(
        question=question,
        provider=provider,
        model=model or None,
        api_key=api_key or None,
        base_url=base_url or None,
        session_id=session_id or None,
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")