damascus-orchestrator/scripts/_verify_mcp_helper.py

"""Damascus MCP stdio helper for scripts/verify.sh.

Drives ``python -m damascus.mcp_server`` over stdio via the official
``mcp`` SDK client. The MCP server is a thin wrapper around
``damascus-api`` (loopback HTTP); this helper just frames the JSON-RPC
for the bash wrapper script so the bash doesn't have to manage
heredocs, Content-Length headers, or mcp SDK imports.

Subcommands
-----------

``initialize``
    Send the MCP ``initialize`` handshake; print server name + version
    as a single JSON line on stdout.

``list-tools``
    Send ``tools/list`` after the handshake; print the sorted tool
    name list + count as a single JSON line.

``ingest-story PROJECT STORY_ID TITLE PRIORITY``
    Call ``tools/call ingest_story`` and print
    ``{"server_name": ..., "payload": <API response>}``.

Auth
----
The helper reads ``DAMASCUS_API_TOKEN`` from the shell env, falling back
to ``/root/.hermes/.env`` (the same source ``damascus-api`` itself
reads). The MCP process is launched via ``docker compose exec
damascus-api python -m damascus.mcp_server`` and inherits ``DAMASCUS_API_BASE=http://damascus-api:9110`` so the container DNS
resolves the upstream.

Exit codes
----------
``0`` on success, ``1`` on a runtime error, ``2`` on bad arguments.
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import sys
from pathlib import Path

from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

# Silence the SDK's "Tool <name> not listed, no validation will be
# performed" warning emitted on every call_tool. The MCP server declares
# `ingest_story` in its catalog but the SDK's structured-output validator
# still complains because the server does not return a `structuredContent`
# block (it returns the API payload as TextContent). Validation is
# not actionable here — the bash wrapper asserts the JSON shape itself.
logging.getLogger("mcp.client.session").setLevel(logging.ERROR)


ENV_FILE = Path("/root/.hermes/.env")
COMPOSE_FILE = "/root/damascus-orchestrator/docker-compose.yml"
TOKEN_KEY = "DAMASCUS_API_TOKEN"


def _load_token() -> str:
    token = os.environ.get(TOKEN_KEY, "").strip()
    if token:
        return token
    if not ENV_FILE.exists():
        return ""
    for raw in ENV_FILE.read_text().splitlines():
        line = raw.strip()
        if line.startswith("export "):
            line = line[len("export "):].lstrip()
        if not line.startswith(TOKEN_KEY + "="):
            continue
        val = line.split("=", 1)[1].strip()
        if (val.startswith("'") and val.endswith("'")) or (
            val.startswith('"') and val.endswith('"')
        ):
            val = val[1:-1]
        return val
    return ""


def _stdio_params() -> StdioServerParameters:
    token = _load_token()
    if not token:
        print(f"[verify-mcp] {TOKEN_KEY} not found in env or {ENV_FILE}", file=sys.stderr)
        sys.exit(2)
    # The MCP process runs inside damascus-api (via `docker compose exec`),
    # so it needs the container-DNS upstream URL — not localhost:9110.
    api_base = os.environ.get("DAMASCUS_API_BASE_FOR_MCP", "http://damascus-api:9110")
    return StdioServerParameters(
        command="docker",
        args=[
            "compose",
            "-f",
            COMPOSE_FILE,
            "exec",
            "-T",
            "damascus-api",
            "python",
            "-m",
            "damascus.mcp_server",
        ],
        env={
            **os.environ,
            "DAMASCUS_API_BASE": api_base,
            TOKEN_KEY: token,
        },
    )


async def _run(sub: str, rest: list[str]) -> int:
    params = _stdio_params()
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            init = await session.initialize()
            server_name = init.serverInfo.name

            if sub == "initialize":
                print(json.dumps({
                    "server_name": server_name,
                    "server_version": init.serverInfo.version,
                }))
                return 0

            if sub == "list-tools":
                tools = await session.list_tools()
                names = sorted(t.name for t in tools.tools)
                print(json.dumps({
                    "server_name": server_name,
                    "tool_names": names,
                    "tool_count": len(names),
                }))
                return 0

            if sub == "ingest-story":
                if len(rest) < 4:
                    print(
                        "[verify-mcp] ingest-story requires "
                        "PROJECT STORY_ID TITLE PRIORITY",
                        file=sys.stderr,
                    )
                    return 2
                project, story_id, title, priority = rest[:4]
                res = await session.call_tool(
                    "ingest_story",
                    arguments={
                        "project": project,
                        "story_id": story_id,
                        "title": title,
                        "priority": int(priority),
                    },
                )
                if not res.content:
                    print("[verify-mcp] empty content from ingest_story", file=sys.stderr)
                    return 1
                payload = json.loads(res.content[0].text)
                print(json.dumps({"server_name": server_name, "payload": payload}))
                return 0

            print(f"[verify-mcp] unknown subcommand: {sub!r}", file=sys.stderr)
            return 2


def main() -> int:
    if len(sys.argv) < 2:
        print(__doc__, file=sys.stderr)
        return 2
    sub = sys.argv[1]
    rest = sys.argv[2:]
    try:
        return asyncio.run(_run(sub, rest))
    except Exception as exc:
        print(f"[verify-mcp] {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())