orchestrata_ultimate.tools.mcp_stdio_client

Reference › orchestrata_ultimate › tools › mcp_stdio_client

This module is a minimal client for the Model Context Protocol over stdio. It is used only by mcp_resource_tools, and tests/tools/test_mcp_resource_tools.py covers it.

McpStdioClient takes a command, its arguments and a timeout (default 10 seconds) and speaks JSON-RPC 2.0 over the child process's stdin and stdout. It is an async context manager. connect starts the subprocess with piped streams and performs the initialize handshake, then sends the initialized notification and keeps the server info. list_resources calls resources/list and returns McpResourceEntry records. read_resource calls resources/read and returns McpResourceContent items holding text or a base64 blob. close closes stdin and waits for the server, killing it if it hangs.

_receive reads one line at a time under the timeout and skips notifications and responses with other ids until it finds the expected id. Failures raise McpClientError, which carries an error_code and optional JSON-RPC code: SERVER_UNREACHABLE when the process cannot start or closes its output (with up to 500 characters of stderr), READ_TIMEOUT on timeout, and SERVER_ERROR for invalid JSON or an error reply.

At a glance

Class diagram

Dependency diagram

API

Classes

McpClientError

Bases: Exception · Line 41

Raised for MCP stdio transport/protocol failures.

McpResourceEntry (dataclass)

Decorators: dataclass · Line 65

One entry from a resources/list response.

McpResourceContent (dataclass)

Decorators: dataclass · Line 76

One content item from a resources/read response.

McpStdioClient

Line 89

Speaks JSON-RPC 2.0 over a child process's stdin/stdout (MCP stdio transport).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/mcp_stdio_client.py

"""McpStdioClient — a minimal JSON-RPC-over-stdio client for MCP servers.

Spec: SPEC-071 (MCP Resource Tools). spec.md's "Runtime Dependencies" table
names the official ``mcp`` Python SDK (>=1.0.0) as the intended transport —
**not declared** in ``pyproject.toml`` (verified: ``grep -i mcp pyproject.toml``
finds nothing). Rather than hand-roll the SDK's full surface (sampling,
prompts, tool-calling, SSE/HTTP transports, capability negotiation) or add an
undeclared dependency, this module implements exactly the subset SPEC-071's
two in-scope tools need: launch a server subprocess, complete the
``initialize`` handshake, and call ``resources/list`` / ``resources/read``
over the stdio transport's newline-delimited JSON-RPC 2.0 framing.

Spec fork — report to the orchestrator: this is intentionally the
production-hardening seam. A follow-up spec/PR should swap this hand-written
client for the official ``mcp`` SDK's ``ClientSession`` + ``stdio_client``
(``uv add mcp``) once that dependency is approved, and add a persistent
per-server connection pool (SPEC-071's ``MCPConnection``/"Connection Pool
Management") instead of this MVP's connect-per-call model.
"""

from __future__ import annotations

import asyncio
import itertools
import json
from dataclasses import dataclass
from typing import Any

#: MCP protocol version this client negotiates in ``initialize`` (spec.md
#: doesn't pin one; this is the latest stable revision at time of writing).
_PROTOCOL_VERSION = "2024-11-05"

#: JSON-RPC error code this client recognizes as "resource not found"
#: (REQ-071-002/ERR-002). Not part of the base JSON-RPC 2.0 spec — it is a
#: convention some resource-serving MCP servers use for out-of-band
#: application errors; servers that don't use it simply surface as the
#: generic SERVER_ERROR code instead.
_RESOURCE_NOT_FOUND_RPC_CODE = -32001


class McpClientError(Exception):
    """Raised for MCP stdio transport/protocol failures.

    Carries ``error_code`` — one of the SPEC-071 "Error Conditions" table IDs
    (SERVER_UNREACHABLE/READ_TIMEOUT/SERVER_ERROR/RESOURCE_NOT_FOUND/...) —
    so tool-layer callers surface a stable code without string-matching this
    exception's message, and ``rpc_code`` — the raw JSON-RPC ``error.code``
    when the failure came from a JSON-RPC error response, or ``None``
    otherwise.
    """

    def __init__(
        self,
        message: str,
        *,
        error_code: str = "SERVER_ERROR",
        rpc_code: int | None = None,
    ) -> None:
        super().__init__(message)
        self.error_code = error_code
        self.rpc_code = rpc_code


@dataclass
class McpResourceEntry:
    """One entry from a ``resources/list`` response."""

    uri: str
    name: str
    mime_type: str | None = None
    size: int | None = None
    description: str | None = None


@dataclass
class McpResourceContent:
    """One content item from a ``resources/read`` response.

    Exactly one of ``text``/``blob`` is populated, per the MCP resource
    content shape (``blob`` is base64-encoded binary content).
    """

    uri: str
    mime_type: str | None = None
    text: str | None = None
    blob: str | None = None


class McpStdioClient:
    """Speaks JSON-RPC 2.0 over a child process's stdin/stdout (MCP stdio transport).

    MVP scope (see module docstring "Spec fork"): one subprocess per
    connect/close cycle — no connection pooling, no server registry, no
    auth/vault/cache layers. Use as an async context manager::

        async with McpStdioClient("python", ["server.py"]) as client:
            resources = await client.list_resources()
            contents = await client.read_resource(resources[0].uri)
    """

    def __init__(
        self,
        command: str,
        args: list[str] | None = None,
        *,
        timeout: float = 10.0,
    ) -> None:
        self._command = command
        self._args = list(args or [])
        self._timeout = timeout
        self._proc: asyncio.subprocess.Process | None = None
        self._id_counter = itertools.count(1)
        #: Populated from the ``initialize`` response's ``serverInfo`` field
        #: once ``connect()`` completes.
        self.server_info: dict[str, Any] = {}

    async def __aenter__(self) -> McpStdioClient:
        await self.connect()
        return self

    async def __aexit__(self, *exc_info: object) -> None:
        await self.close()

    # ── lifecycle ────────────────────────────────────────────────────

    async def connect(self) -> None:
        """Launch the server subprocess and complete the ``initialize`` handshake."""
        try:
            self._proc = await asyncio.create_subprocess_exec(
                self._command,
                *self._args,
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
        except OSError as exc:
            raise McpClientError(
                f"could not launch MCP server {self._command!r}: {exc}",
                error_code="SERVER_UNREACHABLE",
            ) from exc

        init_result = await self._request(
            "initialize",
            {
                "protocolVersion": _PROTOCOL_VERSION,
                "capabilities": {},
                "clientInfo": {"name": "orchestrata-ultimate", "version": "1.0.0"},
            },
        )
        self.server_info = init_result.get("serverInfo", {}) or {}
        await self._notify("notifications/initialized", {})

    async def close(self) -> None:
        """Close stdin and wait for the server to exit, killing it if it hangs."""
        proc = self._proc
        if proc is None:
            return
        self._proc = None
        if proc.stdin is not None and not proc.stdin.is_closing():
            proc.stdin.close()
        try:
            await asyncio.wait_for(proc.wait(), timeout=self._timeout)
        except TimeoutError:
            proc.kill()
            await proc.wait()

    # ── resource operations ──────────────────────────────────────────

    async def list_resources(self) -> list[McpResourceEntry]:
        """Call ``resources/list`` and return the parsed resource entries."""
        result = await self._request("resources/list", {})
        raw_resources = result.get("resources", [])
        return [
            McpResourceEntry(
                uri=r["uri"],
                name=r.get("name", r["uri"]),
                mime_type=r.get("mimeType"),
                size=r.get("size"),
                description=r.get("description"),
            )
            for r in raw_resources
        ]

    async def read_resource(self, uri: str) -> list[McpResourceContent]:
        """Call ``resources/read`` for ``uri`` and return the parsed content items."""
        try:
            result = await self._request("resources/read", {"uri": uri})
        except McpClientError as exc:
            if exc.rpc_code == _RESOURCE_NOT_FOUND_RPC_CODE:
                raise McpClientError(str(exc), error_code="RESOURCE_NOT_FOUND") from exc
            raise
        raw_contents = result.get("contents", [])
        return [
            McpResourceContent(
                uri=c.get("uri", uri),
                mime_type=c.get("mimeType"),
                text=c.get("text"),
                blob=c.get("blob"),
            )
            for c in raw_contents
        ]

    # ── JSON-RPC transport ───────────────────────────────────────────

    async def _request(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        if self._proc is None or self._proc.stdin is None or self._proc.stdout is None:
            raise McpClientError(
                "client is not connected", error_code="SERVER_UNREACHABLE"
            )
        req_id = next(self._id_counter)
        await self._send(
            {"jsonrpc": "2.0", "id": req_id, "method": method, "params": params}
        )
        return await self._receive(req_id)

    async def _notify(self, method: str, params: dict[str, Any]) -> None:
        if self._proc is None or self._proc.stdin is None:
            raise McpClientError(
                "client is not connected", error_code="SERVER_UNREACHABLE"
            )
        await self._send({"jsonrpc": "2.0", "method": method, "params": params})

    async def _send(self, message: dict[str, Any]) -> None:
        assert self._proc is not None and self._proc.stdin is not None
        line = json.dumps(message) + "\n"
        self._proc.stdin.write(line.encode("utf-8"))
        await self._proc.stdin.drain()

    async def _receive(self, expected_id: int) -> dict[str, Any]:
        assert self._proc is not None and self._proc.stdout is not None
        proc = self._proc
        assert proc.stdout is not None
        while True:
            try:
                raw = await asyncio.wait_for(
                    proc.stdout.readline(), timeout=self._timeout
                )
            except TimeoutError as exc:
                raise McpClientError(
                    f"timed out waiting for a response to request id={expected_id}",
                    error_code="READ_TIMEOUT",
                ) from exc
            if not raw:
                stderr = b""
                if proc.stderr is not None:
                    stderr = await proc.stderr.read()
                raise McpClientError(
                    "MCP server closed its stdout unexpectedly "
                    f"(stderr: {stderr.decode('utf-8', 'replace')[:500]!r})",
                    error_code="SERVER_UNREACHABLE",
                )
            line = raw.decode("utf-8").strip()
            if not line:
                continue
            try:
                payload = json.loads(line)
            except json.JSONDecodeError as exc:
                raise McpClientError(
                    f"invalid JSON-RPC line from server: {line[:200]!r}",
                    error_code="SERVER_ERROR",
                ) from exc
            if payload.get("id") != expected_id:
                # A notification or a response to an earlier/out-of-order
                # request — not what we're waiting for; keep reading.
                continue
            if "error" in payload:
                err = payload["error"] or {}
                raise McpClientError(
                    f"MCP server error {err.get('code')}: {err.get('message')}",
                    error_code="SERVER_ERROR",
                    rpc_code=err.get("code"),
                )
            return payload.get("result", {}) or {}