claude-code-api/scripts/echo_mcp_server.py

"""Minimal stdio MCP server with one tool: `echo`.

Used by the tool-calling smoke test to exercise the `--mcp-config` path
end-to-end with a real `claude` interactive process. We hand-roll the
JSON-RPC framing rather than pulling in the `mcp` Python SDK so the test
has the same zero-dep posture as the rest of the package.

Protocol surface (Model Context Protocol, stdio transport):
    - `initialize`                 -> capabilities + serverInfo + protocolVersion
    - `notifications/initialized`  -> no response
    - `tools/list`                 -> list of {name, description, inputSchema}
    - `tools/call`                 -> content blocks for the named tool
    - `ping`                       -> empty result

The `echo` tool returns its `text` argument verbatim as a single
`{"type":"text","text":...}` content block. That is enough for claude to
surface a `tool_use` -> `tool_result` pair in the JSONL session file.

Frame format on the wire is the stdio MCP convention: one JSON object per
line on stdin/stdout. We deliberately do NOT speak the alternative
`Content-Length`-framed flavor — claude's `--mcp-config` stdio transport
uses line framing for spawn-and-pipe servers.

Run standalone (for protocol sanity-checking) with:
    printf '%s\\n%s\\n%s\\n' \\
      '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"x","version":"1"}}}' \\
      '{"jsonrpc":"2.0","method":"notifications/initialized"}' \\
      '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"hi"}}}' \\
      | python scripts/echo_mcp_server.py
"""

from __future__ import annotations

import json
import os
import sys
import traceback
from pathlib import Path
from typing import Any

_LOG_PATH = os.environ.get("ECHO_MCP_LOG")


def _log(message: str) -> None:
    if not _LOG_PATH:
        return
    try:
        with Path(_LOG_PATH).open("a", encoding="utf-8") as f:
            f.write(message.rstrip("\n") + "\n")
    except OSError:
        pass


DEFAULT_PROTOCOL_VERSION = "2025-06-18"

SERVER_NAME = "echo-server"
SERVER_VERSION = "0.0.1"

ECHO_TOOL = {
    "name": "echo",
    "description": "Return the supplied `text` argument verbatim.",
    "inputSchema": {
        "type": "object",
        "properties": {
            "text": {"type": "string", "description": "Text to echo back."},
        },
        "required": ["text"],
        "additionalProperties": False,
    },
}


def _write(message: dict[str, Any]) -> None:
    """Emit one JSON-RPC message as a single newline-terminated stdout line."""
    sys.stdout.write(json.dumps(message, ensure_ascii=False) + "\n")
    sys.stdout.flush()


def _ok(req_id: Any, result: dict[str, Any]) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "result": result}


def _err(req_id: Any, code: int, message: str) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}


def _handle_initialize(params: dict[str, Any]) -> dict[str, Any]:
    requested = params.get("protocolVersion") if isinstance(params, dict) else None
    version = requested if isinstance(requested, str) else DEFAULT_PROTOCOL_VERSION
    return {
        "protocolVersion": version,
        "capabilities": {"tools": {"listChanged": False}},
        "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
    }


def _handle_tools_list() -> dict[str, Any]:
    return {"tools": [ECHO_TOOL]}


def _handle_tools_call(params: dict[str, Any]) -> dict[str, Any]:
    name = params.get("name") if isinstance(params, dict) else None
    arguments = params.get("arguments") if isinstance(params, dict) else None
    if name != "echo":
        return {
            "content": [{"type": "text", "text": f"unknown tool: {name!r}"}],
            "isError": True,
        }
    text = ""
    if isinstance(arguments, dict):
        raw = arguments.get("text", "")
        text = raw if isinstance(raw, str) else json.dumps(raw)
    return {"content": [{"type": "text", "text": text}], "isError": False}


def _dispatch(message: dict[str, Any]) -> dict[str, Any] | None:
    method = message.get("method")
    req_id = message.get("id")
    params = message.get("params") or {}

    is_notification = "id" not in message

    if method == "initialize":
        return _ok(req_id, _handle_initialize(params))
    if method == "notifications/initialized":
        return None
    if method == "tools/list":
        return _ok(req_id, _handle_tools_list())
    if method == "tools/call":
        return _ok(req_id, _handle_tools_call(params))
    if method == "ping":
        return _ok(req_id, {})
    if method == "prompts/list":
        return _ok(req_id, {"prompts": []})
    if method == "resources/list":
        return _ok(req_id, {"resources": []})
    if method == "resources/templates/list":
        return _ok(req_id, {"resourceTemplates": []})

    if is_notification:
        return None
    return _err(req_id, -32601, f"method not found: {method!r}")


def main() -> int:
    _log(f"START pid={os.getpid()} python={sys.executable} cwd={Path.cwd()}")
    try:
        while True:
            raw_line = sys.stdin.readline()
            if not raw_line:
                _log("EOF on stdin")
                break
            line = raw_line.strip()
            if not line:
                continue
            _log(f"RX {line[:2000]}")
            try:
                message = json.loads(line)
            except json.JSONDecodeError as exc:
                _log(f"parse error: {exc}")
                _write(_err(None, -32700, f"parse error: {exc}"))
                continue
            if not isinstance(message, dict):
                _write(_err(None, -32600, "invalid request: not a JSON object"))
                continue
            response = _dispatch(message)
            if response is not None:
                _log(f"TX {json.dumps(response)[:2000]}")
                _write(response)
    except Exception as exc:
        _log("CRASH: " + "".join(traceback.format_exception(exc)))
        raise
    _log("EXIT 0")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())