stdio_server.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """MuseHub MCP stdio transport — local dev and Cursor integration. |
| 2 | |
| 3 | Entry point: ``python -m musehub.mcp.stdio_server`` |
| 4 | |
| 5 | Protocol: |
| 6 | - Reads newline-delimited JSON-RPC 2.0 messages from ``stdin`` |
| 7 | - Writes JSON-RPC 2.0 responses to ``stdout`` (one per line) |
| 8 | - Logs diagnostic messages to ``stderr`` |
| 9 | - Processes messages sequentially (no concurrency concerns) |
| 10 | - Notifications (no ``id``) produce no output (202-equivalent silence) |
| 11 | |
| 12 | Auth: |
| 13 | - No auth required — stdio is a trusted local process. |
| 14 | - All tools including write tools are available. |
| 15 | - ``user_id`` defaults to ``"stdio-user"`` so write tools can record an author. |
| 16 | |
| 17 | Usage: |
| 18 | # Direct: |
| 19 | python -m musehub.mcp.stdio_server |
| 20 | |
| 21 | # Via Cursor (.cursor/mcp.json): |
| 22 | { |
| 23 | "mcpServers": { |
| 24 | "musehub": { |
| 25 | "command": "python", |
| 26 | "args": ["-m", "musehub.mcp.stdio_server"], |
| 27 | "env": {"DATABASE_URL": "postgresql+asyncpg://..."} |
| 28 | } |
| 29 | } |
| 30 | } |
| 31 | """ |
| 32 | |
| 33 | import asyncio |
| 34 | import json |
| 35 | import logging |
| 36 | import sys |
| 37 | |
| 38 | from musehub.types.json_types import JSONObject |
| 39 | |
| 40 | logger = logging.getLogger("musehub.mcp.stdio") |
| 41 | |
| 42 | |
| 43 | async def _run() -> None: |
| 44 | """Main async stdio loop.""" |
| 45 | from musehub.db import init_db |
| 46 | from musehub.mcp.dispatcher import handle_request |
| 47 | from musehub.mcp.session import MCPSession, create_session_from_initialize_params |
| 48 | |
| 49 | # Initialise DB before handling any requests. echo=False is forced |
| 50 | # regardless of settings.debug -- SQLAlchemy's echo=True installs its |
| 51 | # own hardcoded sys.stdout log handler (musehub#208), which would |
| 52 | # corrupt the JSON-RPC wire protocol this transport writes to stdout. |
| 53 | try: |
| 54 | await init_db(echo=False) |
| 55 | logger.info("MuseHub MCP stdio server started") |
| 56 | except Exception as exc: |
| 57 | logger.warning("DB init failed (tools requiring DB will return db_unavailable): %s", exc) |
| 58 | |
| 59 | loop = asyncio.get_event_loop() |
| 60 | reader = asyncio.StreamReader() |
| 61 | protocol = asyncio.StreamReaderProtocol(reader) |
| 62 | await loop.connect_read_pipe(lambda: protocol, sys.stdin) |
| 63 | |
| 64 | stdout_transport, stdout_protocol = await loop.connect_write_pipe( |
| 65 | asyncio.BaseProtocol, sys.stdout |
| 66 | ) |
| 67 | |
| 68 | # musehub#210: stdio is a single, long-lived connection to exactly one |
| 69 | # client for the lifetime of this process -- unlike the Streamable HTTP |
| 70 | # transport, there is no Mcp-Session-Id header to key a session lookup |
| 71 | # by. Without a session object at all, musehub_set_context's write to |
| 72 | # session.repo_focus never happened (guarded by `if ctx.session is not |
| 73 | # None`), so the very next musehub_read_context() call -- with no |
| 74 | # session to inject owner/slug from -- resolved to "". Create one |
| 75 | # session right after a successful `initialize`, the same moment the |
| 76 | # HTTP transport does, and reuse it for every request thereafter. |
| 77 | session: MCPSession | None = None |
| 78 | |
| 79 | while True: |
| 80 | try: |
| 81 | line_bytes = await reader.readline() |
| 82 | except asyncio.IncompleteReadError: |
| 83 | break |
| 84 | if not line_bytes: |
| 85 | break |
| 86 | |
| 87 | line = line_bytes.decode("utf-8").strip() |
| 88 | if not line: |
| 89 | continue |
| 90 | |
| 91 | try: |
| 92 | raw = json.loads(line) |
| 93 | except json.JSONDecodeError as exc: |
| 94 | error_resp = { |
| 95 | "jsonrpc": "2.0", |
| 96 | "id": None, |
| 97 | "error": {"code": -32700, "message": f"Parse error: {exc}"}, |
| 98 | } |
| 99 | _write_json(stdout_transport, error_resp) |
| 100 | continue |
| 101 | |
| 102 | try: |
| 103 | resp = await handle_request(raw, user_id="stdio-user", session=session) |
| 104 | except Exception as exc: |
| 105 | logger.exception("Dispatcher error: %s", exc) |
| 106 | resp = { |
| 107 | "jsonrpc": "2.0", |
| 108 | "id": raw.get("id"), |
| 109 | "error": {"code": -32603, "message": f"Internal error: {exc}"}, |
| 110 | } |
| 111 | |
| 112 | if resp is not None and raw.get("method") == "initialize" and "error" not in resp: |
| 113 | from musehub.mcp.session import SessionCapacityError |
| 114 | try: |
| 115 | session = create_session_from_initialize_params(raw, "stdio-user") |
| 116 | except SessionCapacityError as exc: |
| 117 | logger.warning( |
| 118 | "Could not create stdio MCP session (%s) — " |
| 119 | "musehub_set_context session focus will not persist.", |
| 120 | exc, |
| 121 | ) |
| 122 | |
| 123 | if resp is not None: |
| 124 | _write_json(stdout_transport, resp) |
| 125 | |
| 126 | |
| 127 | def _write_json(transport: asyncio.BaseTransport, obj: JSONObject) -> None: |
| 128 | """Serialise ``obj`` to JSON and write it as a single newline-terminated line.""" |
| 129 | line = f"{json.dumps(obj, default=str)}\n" |
| 130 | if isinstance(transport, asyncio.WriteTransport): |
| 131 | transport.write(line.encode("utf-8")) |
| 132 | |
| 133 | |
| 134 | def main() -> None: |
| 135 | """Stdio server entry point.""" |
| 136 | logging.basicConfig( |
| 137 | level=logging.INFO, |
| 138 | format="%(asctime)s %(levelname)s %(name)s: %(message)s", |
| 139 | stream=sys.stderr, |
| 140 | ) |
| 141 | asyncio.run(_run()) |
| 142 | |
| 143 | |
| 144 | if __name__ == "__main__": |
| 145 | main() |