docs: document MCP server config, runtime, and /mcp command

CLAUDE.md

@@ -56,12 +56,14 @@ lmproto/deepseek/anthropic/     ← DeepSeek Anthropic client (extends Anthropic
 
 ### Config (`config/`)
 
-TOML load/store (`ConfigStore`): `[providers]` tagged union presets, `[database]` section. Default path via platformdirs. Providers may set `timeout` as a float or `{ connect, read }` (`HttpTimeoutConfig`); omitted → product defaults (10s connect / 600s read). `DeepseekProvider.convention` (Literal: `""`/`"openai"`/`"anthropic"`/`"responses"`) selects the DeepSeek API surface — `"responses"` → OpenAI Responses, `"anthropic"` → Anthropic Messages on `https://api.deepseek.com/anthropic` (default url follows the convention); a per-model `ModelConfig.convention` overrides it (empty = inherit).
+TOML load/store (`ConfigStore`): `[providers]` tagged union presets, `[database]` section. Default path via platformdirs. Providers may set `timeout` as a float or `{ connect, read }` (`HttpTimeoutConfig`); omitted → product defaults (10s connect / 600s read). `DeepseekProvider.convention` (Literal: `""`/`"openai"`/`"anthropic"`/`"responses"`) selects the DeepSeek API surface — `"responses"` → OpenAI Responses, `"anthropic"` → Anthropic Messages on `https://api.deepseek.com/anthropic` (default url follows the convention); a per-model `ModelConfig.convention` overrides it (empty = inherit). `[mcp]` (`McpConfig`): `servers.<name>` stdio server definitions (`McpServerConfig`: `command`/`args`/`env`/`cwd`/`timeout`/`read_only`, `url` reserved for streamable HTTP) + `disable` gate; defining a server opts it in.
 
 ### Runtime (`runtime/`)
 
 `create_client(provider)` maps config `Provider` → protocol client by effective preset (model-level `preset`/`url` overrides apply). `normalize_http_timeout` feeds OpenAI-compatible configs and Anthropic. Mapping: `openai` → `OpenAIClient` (agent uses Responses), `openai-compatible` / `deepseek` → chat-completions clients (deepseek with effective `convention = "responses"` → `DeepseekResponsesClient`, `"anthropic"` → `DeepseekAnthropicClient`), `anthropic` → `AnthropicClient`.
 
+MCP (`runtime/mcp_client.py`): `McpManager` owns one `McpServerConnection` per `[mcp]` server — a spawned stdio subprocess speaking newline-delimited JSON-RPC 2.0: initialize handshake, `tools/list` (cursor pagination), `tools/call`, `notifications/tools/list_changed` → `tools_stale`. Spawn/handshake failures record `connection.error` (status string), never raise; `ensure_started` / `restart(config)` / `aclose` / `refresh_tools`; `statuses()` → `(server, status, tool_count)` rows; per-request timeout from config (default 30s).
+
 ### Memory (`memory/`)
 
 Async SQLAlchemy + aiosqlite. `MemoryStore`: schema init (+ lightweight SQLite `ALTER` for new columns), default local user, sessions (bound to `workspace` path; optional `provider_name`/`model`), messages stored as msgspec chat message JSON.
@@ -94,6 +96,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use.
 - **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core); `wait` (line prompt with timeout; Enter disturbs).
 - **`todo`**: LIFO stack of **task groups** (not a queue of tasks) — `todo_push` creates one group of siblings; `todo_pop` removes the whole top group; `todo_update` by id; DFS breakdown push[T1,T2]→push[T1.1…]→pop→push[T2.1…]; stored on session row; open items = unfinished work (`[TODO OPEN WORK]`); all-terminal non-empty = hygiene (`[TODO HYGIENE]`); nag channel via `[agent] todo_nag_strategy` (`developer`/`user` = prose nag; `synthetic_tool` = forged `todo_list` call + **real** `stack.render()` result + ToolCall/Result events; `none`).
 - **`net` / `fetch`**: HTTP GET/POST/PUT/DELETE via niquests (isolated session, not the LLM client). Manual redirects with per-hop SSRF checks (`asyncio` `getaddrinfo`). Private/loopback hosts need instance policy grant (CLI timed confirm; **not** YOLO). Public cleartext HTTP and mutating methods soft-confirm (`YOLO`/`TRUSTABLE`). HTTPS→HTTP redirects blocked unless `allow_http_downgrade`. `user_agent` tool arg (or headers) is never replaced by the default when provided. Caps: body bytes/chars, request body size.
+- **`mcp`** (`tools/mcp.py`): register one server's advertised tools into the catalog as namespaced `mcp__<server>__<tool>` definitions with `ToolSource(kind="mcp", plugin_id=server)`; tags LOCAL (+ READ_ONLY when the server config sets `read_only`) — never YOLO/TRUSTABLE. `register_mcp_tools` purges prior MCP entries then re-registers every connected server (hosts call it on tool-registry rebuild); connections must already be started by `McpManager`.
 - **`DEFAULT_TOOLS` / catalog**: file + process + vcs + chat + todo + net (`fetch`) for local surface. `ToolTag.READ_ONLY` marks non-mutating tools (`read_file`, `get_truncated`, `listdir`, `glob_paths`, `grep_files`, `tree`, `todo_list`, `wait`, `ask_user_*`, `vcs_*`, `fetch`); `ToolRegistry.clone(read_only_only=True)` selects just those for safe side turns / planning, and `read_only_context()` (in `tools/context.py`) makes `fetch` accept GET only.
 
 ### Prompting (`prompting.py`)
@@ -104,8 +107,8 @@ Shared interactive I/O: `ask` / `choose` / `form` / `confirm` with pluggable bac
 
 Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
 
-- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`.
-- Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack.
+- **`plyngent chat`**: provider/model (flags or interactive; Tab via readline in `prompting`); sessions store `provider_name`/`model` and restore on resume; SQLite via `[database]` (file DB under user data when url unset; explicit `url = ":memory:"` kept + warn); workspace-bound; resume latest for cwd/`--workspace` by default (`--new` / `--session`). One-shot: `-p/--prompt` and non-TTY stdin; exit codes 0/1/2/3; `--yes` (YOLO on), `--stream/--no-stream`, `--quiet`. Root `--log-level`. Enabled `[mcp]` servers start before the REPL (`_start_mcp_manager`, per-server failures warn on stderr) and are closed on exit (`McpManager.aclose`).
+- Slash: Click group in `cli/slash.py` + `awaitlet` for async work; Tab completer from registry + ParamType `shell_complete`. Multiline `"""` … `"""`; `/edit` via `$VISUAL`/`$EDITOR` (blocking only). `/yolo on|off|once` for soft destructive confirms. `/model --persist` / `/models --persist` write model catalog entries into TOML. `/todos` for human show/push/pop/clear of the todo stack. `/mcp` (`list` default) prints per-server status/tool counts + last stderr lines; `/mcp reconnect` re-reads config, restarts every enabled server, and rebuilds the tool registry.
 - `/btw` side questions: forked session; main transcript/DB untouched. `--tools=read` (default) exposes only `READ_ONLY` tools under `read_only_context()` (fetch=GET) with the main session view; `--tools=no` off; `--tools=full` whole-registry clone with fresh session. Tools off: default falls back to no tools; explicit `read`/`full` errors.
 - Explicit `/resume` or `--session` from another workspace prompts: **keep** / **update** / **abort**.
 - Failed/cancelled turns: user kept; **committed tool rounds kept** (side effects not re-run on `/retry`); only unfinished assistant rolled back; Ctrl+C cancels; TTY confirms off-loop; auto-retry 10s/20s/30s then `/retry`.

README.md

@@ -157,6 +157,8 @@ Per-provider **`timeout`** is passed to the HTTP session for chat/completions, R
 
 Third-party **plugins**: install a package that declares `project.entry-points."plyngent.tools"` (and later other groups), then allowlist the entry-point name under **`[plugins].enable`**. Details: [doc/plugins.md](doc/plugins.md).
 
+**MCP servers** (Model Context Protocol over stdio): define servers under `[mcp.servers.<name>]` (`command` + `args`, optional `env`/`cwd`/`timeout`/`read_only`) and their tools become agent tools namespaced `mcp__<server>__<tool>` (LOCAL tags; `read_only = true` also marks them READ_ONLY, eligible for `/btw --tools=read`). Names in `[mcp].disable` stay disconnected. In the REPL, `/mcp` lists per-server status and tool counts; `/mcp reconnect` re-reads the config file and restarts every enabled server (adopts newly added ones).
+
 Supported provider presets today:
 
 | Preset | API used by the agent | Notes |

doc/plyngent.example.toml

@@ -26,6 +26,19 @@
 # [networking]
 # ssrf_assume_public_cidrs = ["198.18.0.0/15"]
 
+# MCP (Model Context Protocol) servers exposed as agent tools.
+# Defining a server opts it in; tools appear as mcp__<server>__<tool> with
+# LOCAL tags (confirm flow applies). Names in disable stay disconnected.
+# [mcp.servers.docs]
+# command = "npx"
+# args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
+# env = { RUST_LOG = "info" }
+# cwd = "/path"              # spawn working directory (empty = inherit)
+# timeout = 30.0             # per-request timeout (seconds)
+# read_only = true           # mark every tool READ_ONLY (eligible for /btw --tools=read)
+# [mcp]
+# disable = ["docs"]         # defined but never connected
+
 [agent]
 # Persona (omit → built-in coding-agent line). tool_directives is the tool playbook.
 # Override with multi-line literals (prefer ''' so nested " is fine):

src/plyngent/cli/app.py

@@ -26,7 +26,7 @@ from plyngent.cli.state import ReplState
 from plyngent.config.models import DatabaseConfig
 from plyngent.memory import MemoryStore
 from plyngent.prompting import NonInteractiveBackend, configure_prompting
-from plyngent.runtime import ProviderNotSupportedError, create_client
+from plyngent.runtime import McpManager, ProviderNotSupportedError, create_client
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -134,6 +134,24 @@ def _setup_hooks(*, interactive: bool) -> None:
         PtyManager.set_limit_continue_hook(None)
 
 
+async def _start_mcp_manager(store: ConfigStore) -> McpManager | None:
+    """Start configured MCP servers before the REPL state is built.
+
+    Returns ``None`` when nothing is enabled. Failures degrade to a warning
+    per server (the rest of the session, including LLM traffic, continues).
+    """
+    manager = McpManager(store.mcp_config)
+    if not manager.enabled_servers():
+        return None
+    await manager.ensure_started()
+    for name, status, count in manager.statuses():
+        if status.startswith("error"):
+            click.secho(f"warning: mcp server {name!r} failed: {status}", fg="yellow", err=True)
+        else:
+            click.secho(f"mcp: {name} ({status}, tools={count})", fg="bright_black", err=True)
+    return manager
+
+
 async def _bind_session(
     state: ReplState,
     *,
@@ -285,6 +303,7 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
         except ProviderNotSupportedError as exc:
             raise click.ClickException(str(exc)) from exc
 
+        mcp_manager = await _start_mcp_manager(store)
         state = ReplState(
             config=store,
             memory=memory,
@@ -297,6 +316,7 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             stream_enabled=stream,
             interactive_limits=interactive,
             yolo=yolo,
+            mcp_manager=mcp_manager,
         )
         # Path denylist and policy confirm live on instance.workspace (no process bag).
         state.instance_state.workspace.path_denylist = tuple(store.agent_config.path_denylist or ())
@@ -353,6 +373,8 @@ async def _run_chat(  # noqa: C901, PLR0912, PLR0915 — chat orchestration
             set_fetch_policy_confirm_hook(None, instance=state_obj.instance_state)
             clear_private_grants(instance=state_obj.instance_state)
             clear_ssrf_assume_public_cidrs()
+            if state_obj.mcp_manager is not None:
+                await state_obj.mcp_manager.aclose()
             await state_obj.instance_state.shutdown()
         else:
             # No ReplState: only PTY class cleanup (temps require instance allowlist).

src/plyngent/cli/slash.py

@@ -1147,6 +1147,53 @@ def plugins_slash_cmd(state: ReplState, action: str | None, name: str | None) ->
         raise click.UsageError(str(exc)) from exc
 
 
+def _print_mcp_status(state: ReplState) -> None:
+    manager = state.mcp_manager
+    if manager is None:
+        click.echo("mcp: no servers configured (or none enabled)")
+        return
+    rows = manager.statuses()
+    if not rows:
+        click.echo("mcp: no servers configured (or none enabled)")
+        return
+    for name, status, count in rows:
+        click.echo(f"  {name}\t{status}\ttools={count}")
+        connection = next(
+            (conn for conn in manager.connections() if conn.name == name),
+            None,
+        )
+        if connection is not None and connection.error is not None:
+            for line in connection.stderr_tail()[-3:]:
+                click.secho(f"    {line}", fg="bright_black")
+
+
+@slash.command("mcp")
+@click.argument("action", required=False, type=click.Choice(["list", "reconnect"]))
+@click.pass_obj
+def mcp_slash_cmd(state: ReplState, action: str | None) -> None:
+    """List MCP server statuses, or reconnect after config edits.
+
+    ``/mcp`` or ``/mcp list`` — per-server connection status and tool count.
+    ``/mcp reconnect`` — re-read the config file, restart every enabled
+    server, and rebuild the tool registry (adopts newly added servers too).
+    """
+    act = (action or "list").lower()
+    if act == "list":
+        _print_mcp_status(state)
+        return
+    from plyngent.runtime import McpManager
+
+    state.config.reload()
+    if state.mcp_manager is None:
+        state.mcp_manager = McpManager(state.config.mcp_config)
+    else:
+        _await(state.mcp_manager.restart(state.config.mcp_config))
+    _await(state.mcp_manager.ensure_started())
+    _slash_rebuild_tools_if_on(state)
+    click.echo(f"mcp reconnected; tools={'on' if state.tools_enabled else 'off'}")
+    _print_mcp_status(state)
+
+
 @slash.command("yolo")
 @click.argument("mode", required=False, type=YOLO_MODE, metavar="[on|off|once]")
 @click.pass_obj

src/plyngent/cli/state.py

@@ -32,6 +32,7 @@ if TYPE_CHECKING:
     from plyngent.config.store import ConfigStore
     from plyngent.memory import MemoryStore
     from plyngent.memory.database.schema import Session as SessionRow
+    from plyngent.runtime.mcp_client import McpManager
 
 type YoloMode = Literal["off", "on", "once"]
 
@@ -65,6 +66,8 @@ class ReplState:
     todo_stack: TodoStack = field(default_factory=TodoStack)
     instance_state: InstanceState = field(default_factory=InstanceState)
     session_state: SessionState = field(default_factory=SessionState)
+    # MCP connections started by the host app before this state was built.
+    mcp_manager: McpManager | None = None
     _todo_persist_tasks: set[object] = field(default_factory=set, init=False, repr=False)
     # Session ids for Tab complete (updated when listing/creating/resuming).
     _session_id_cache: list[int] = field(default_factory=list, init=False, repr=False)
@@ -215,6 +218,11 @@ class ReplState:
             plugins_cfg.enable,
             disable=plugins_cfg.disable,
         )
+        if self.mcp_manager is not None:
+            from plyngent.tools.mcp import register_mcp_tools
+
+            # Connections were started by the host; registration is sync-only.
+            _ = register_mcp_tools(self.mcp_manager, self.config.mcp_config)
         # Local surface: builtins + allowlisted plugins (import registers into catalog).
         tools = catalog.select(surface="local")
         yolo = self.effective_yolo() != "off"

src/plyngent/config/__init__.py

@@ -13,6 +13,8 @@ from .models import DatabaseConfig as DatabaseConfig
 from .models import DeepSeekConvention as DeepSeekConvention
 from .models import DeepseekProvider as DeepseekProvider
 from .models import HttpTimeoutConfig as HttpTimeoutConfig
+from .models import McpConfig as McpConfig
+from .models import McpServerConfig as McpServerConfig
 from .models import ModelConfig as ModelConfig
 from .models import NetworkingConfig as NetworkingConfig
 from .models import OpenAICompatibleProvider as OpenAICompatibleProvider

src/plyngent/config/models.py

@@ -143,6 +143,44 @@ class PluginsConfig(Struct, omit_defaults=True):
     disable: list[str] = field(default_factory=list)
 
 
+class McpServerConfig(Struct, omit_defaults=True):
+    """One MCP server definition (stdio transport today).
+
+    TOML example::
+
+        [mcp.servers.docs]
+        command = "npx"
+        args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
+        env = { RUST_LOG = "info" }
+        # cwd = "/path"            # spawn working directory (empty = inherit)
+        # timeout = 30.0            # per-request timeout in seconds
+        # read_only = true          # mark every tool READ_ONLY (safe side turns)
+
+    ``url`` is reserved for the streamable-HTTP transport; empty = stdio.
+    A defined server is connected unless its name appears in ``[mcp].disable``.
+    """
+
+    command: str = ""
+    args: list[str] = field(default_factory=list)
+    env: dict[str, str] = field(default_factory=dict)
+    cwd: str = ""
+    timeout: float = 30.0
+    read_only: bool = False
+    url: str = ""
+
+
+class McpConfig(Struct, omit_defaults=True):
+    """Model Context Protocol servers exposed as agent tools.
+
+    Defining a server under ``[mcp.servers.<name>]`` is the opt-in; tools are
+    namespaced ``mcp__<server>__<tool>`` and registered with LOCAL tags. Names
+    in ``disable`` stay disconnected even when defined.
+    """
+
+    servers: dict[str, McpServerConfig] = field(default_factory=dict)
+    disable: list[str] = field(default_factory=list)
+
+
 class NetworkingConfig(Struct, omit_defaults=True):
     """Host-side network policy for tools such as ``fetch``.
 

src/plyngent/config/store.py

@@ -9,6 +9,7 @@ import tomlkit
 from .models import (
     AgentConfig,
     DatabaseConfig,
+    McpConfig,
     ModelConfig,
     NetworkingConfig,
     PluginsConfig,
@@ -56,6 +57,29 @@ def _parse_networking(raw: dict[str, object]) -> NetworkingConfig:
         return NetworkingConfig()
 
 
+def _parse_mcp(raw: dict[str, object]) -> McpConfig:
+    """Parse the ``[mcp]`` section, falling back to defaults.
+
+    Unknown keys (typos like ``cmds``) are rejected the same way as providers:
+    a malformed section silently degrades to "no MCP servers" instead of a
+    half-parsed one.
+    """
+    try:
+        config = msgspec.convert(raw, McpConfig)
+    except msgspec.ValidationError:
+        return McpConfig()
+    known = {"servers", "disable"}
+    if set(raw) - known:
+        return McpConfig()
+    known_server = {"command", "args", "env", "cwd", "timeout", "read_only", "url"}
+    servers_raw = cast("dict[str, object]", raw.get("servers", {}))
+    for name, raw_entry in servers_raw.items():
+        entry = cast("dict[str, object]", raw_entry)
+        if not isinstance(raw_entry, dict) or not name.strip() or set(entry) - known_server:
+            return McpConfig()
+    return config
+
+
 def _parse_providers(
     document: tomlkit.TOMLDocument,
 ) -> tuple[dict[str, Provider], dict[str, object], dict[str, Provider]]:
@@ -125,6 +149,7 @@ class ConfigStore:
     _agent: AgentConfig
     _plugins: PluginsConfig
     _networking: NetworkingConfig
+    _mcp: McpConfig
     _providers: dict[str, Provider]
     _bad_providers: dict[str, object]
     _recoverable_providers: dict[str, Provider]
@@ -137,6 +162,7 @@ class ConfigStore:
         self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
         self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
         self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
+        self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
         self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(document)
 
     @property
@@ -187,6 +213,13 @@ class ConfigStore:
         """Typed networking section (e.g. Fake-IP SSRF exemptions)."""
         return self._networking
 
+    # -- mcp (read-only) --
+
+    @property
+    def mcp_config(self) -> McpConfig:
+        """Typed MCP section (server definitions + disable list)."""
+        return self._mcp
+
     def set_plugins_enable(self, names: Sequence[str]) -> PluginsConfig:
         """Replace the plugin enable list (in-memory; call :meth:`write` to persist)."""
         cleaned = [name.strip() for name in names if name.strip()]
@@ -390,6 +423,7 @@ class ConfigStore:
         self._agent = _parse_agent(cast("dict[str, object]", raw.get("agent", {})))
         self._plugins = _parse_plugins(cast("dict[str, object]", raw.get("plugins", {})))
         self._networking = _parse_networking(cast("dict[str, object]", raw.get("networking", {})))
+        self._mcp = _parse_mcp(cast("dict[str, object]", raw.get("mcp", {})))
         self._providers, self._bad_providers, self._recoverable_providers = _parse_providers(self._document)
 
     # -- internal sync helpers --

src/plyngent/runtime/__init__.py

@@ -1,5 +1,11 @@
+from plyngent.config.models import McpConfig as McpConfig
+
 from .client_factory import InvalidHttpTimeoutError as InvalidHttpTimeoutError
 from .client_factory import ProviderNotSupportedError as ProviderNotSupportedError
 from .client_factory import create_client as create_client
 from .client_factory import normalize_http_timeout as normalize_http_timeout
 from .client_factory import provider_to_openai_config as provider_to_openai_config
+from .mcp_client import McpConnectionError as McpConnectionError
+from .mcp_client import McpManager as McpManager
+from .mcp_client import McpServerConnection as McpServerConnection
+from .mcp_client import McpTool as McpTool

src/plyngent/runtime/mcp_client.py

@@ -0,0 +1,361 @@
+"""MCP client: Model Context Protocol over stdio (newline-delimited JSON-RPC).
+
+One :class:`McpServerConnection` owns one spawned server subprocess and its
+initialize handshake. :class:`McpManager` owns all configured connections for
+the process: start, tool listing, tool calls, reconnect, shutdown.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.metadata
+import json
+import os
+from collections import deque
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, cast
+
+if TYPE_CHECKING:
+    from plyngent.config.models import McpConfig, McpServerConfig
+
+MCP_PROTOCOL_VERSION = "2025-06-18"
+MCP_CLIENT_NAME = "plyngent"
+DEFAULT_MCP_TIMEOUT = 30.0
+_STDERR_TAIL_LINES = 32
+_CLOSE_GRACE_SECONDS = 2.0
+_MAX_TOOL_PAGES = 1024
+
+type Json = dict[str, object]
+
+
+def _json_dict(raw: object) -> Json:
+    """Narrow an untyped JSON payload to a string-keyed object mapping."""
+    return cast("Json", raw) if isinstance(raw, dict) else {}
+
+
+def _json_list(raw: object) -> list[object]:
+    """Narrow an untyped JSON payload to a list of objects."""
+    return cast("list[object]", raw) if isinstance(raw, list) else []
+
+
+def _client_version() -> str:
+    try:
+        return importlib.metadata.version("plyngent")
+    except importlib.metadata.PackageNotFoundError:
+        return "0.0.0"
+
+
+class McpConnectionError(RuntimeError):
+    """Server process failed to spawn, handshake, or answer in time."""
+
+
+@dataclass(frozen=True, slots=True)
+class McpTool:
+    """One tool advertised by a server (``tools/list`` entry)."""
+
+    name: str
+    description: str
+    input_schema: Json
+
+    @classmethod
+    def from_raw(cls, raw: object) -> McpTool:
+        entry = _json_dict(raw)
+        return cls(
+            name=str(entry.get("name", "")),
+            description=str(entry.get("description", "") or ""),
+            input_schema=_json_dict(entry.get("inputSchema")) or {"type": "object"},
+        )
+
+
+def _format_content(result: Json) -> str:
+    """Flatten a ``tools/call`` result payload into model-facing text."""
+    parts: list[str] = []
+    for item in _json_list(result.get("content")):
+        entry = _json_dict(item)
+        kind = str(entry.get("type", ""))
+        if kind == "text":
+            text = entry.get("text")
+            if isinstance(text, str):
+                parts.append(text)
+        elif kind == "resource":
+            parts.append(f"[resource] {json.dumps(entry, ensure_ascii=False)}")
+        else:
+            mime = entry.get("mimeType", "?")
+            parts.append(f"[{kind or 'unknown'}: {mime}]")
+    return "\n".join(parts)
+
+
+class McpServerConnection:
+    """One server subprocess with an active MCP session (stdio transport)."""
+
+    name: str
+    error: str | None
+    tools: list[McpTool]
+    tools_stale: bool
+
+    _config: McpServerConfig
+    _proc: asyncio.subprocess.Process | None
+    _reader_task: asyncio.Task[None] | None
+    _pending: dict[int, asyncio.Future[Json]]
+    _next_id: int
+    _stderr_tail: deque[str]
+
+    def __init__(self, name: str, config: McpServerConfig) -> None:
+        self.name = name
+        self._config = config
+        self._proc = None
+        self._reader_task = None
+        self._pending = {}
+        self._next_id = 1
+        self._stderr_tail = deque(maxlen=_STDERR_TAIL_LINES)
+        self.error = None
+        self.tools = []
+        self.tools_stale = False
+
+    @property
+    def status(self) -> str:
+        if self.error is not None:
+            return f"error: {self.error}"
+        if self._proc is not None:
+            return "connected (tools changed; /mcp reconnect)" if self.tools_stale else "connected"
+        return "not started"
+
+    @property
+    def connected(self) -> bool:
+        return self._proc is not None and self.error is None
+
+    def stderr_tail(self) -> list[str]:
+        return list(self._stderr_tail)
+
+    async def start(self) -> None:
+        """Spawn the server process and run the initialize handshake."""
+        if self._proc is not None:
+            return
+        try:
+            self._proc = await asyncio.create_subprocess_exec(
+                self._config.command,
+                *self._config.args,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                env=os.environ | self._config.env,
+                cwd=self._config.cwd or None,
+            )
+        except (OSError, ValueError) as exc:
+            self.error = f"spawn {self._config.command!r}: {exc}"
+            self._proc = None
+            return
+        self._reader_task = asyncio.create_task(self._read_loop())
+        try:
+            await self._initialize()
+            self.tools = await self.list_tools()
+        except (McpConnectionError, TimeoutError) as exc:
+            self.error = f"initialize: {exc}"
+            await self.aclose()
+
+    async def aclose(self) -> None:
+        """Terminate the server subprocess and fail pending requests."""
+        proc = self._proc
+        self._proc = None
+        if self._reader_task is not None:
+            _ = self._reader_task.cancel()
+            self._reader_task = None
+        if proc is not None and proc.returncode is None:
+            _ = proc.terminate()
+            try:
+                _ = await asyncio.wait_for(proc.wait(), _CLOSE_GRACE_SECONDS)
+            except TimeoutError:
+                proc.kill()
+                _ = await proc.wait()
+        for future in self._pending.values():
+            if not future.done():
+                future.set_exception(McpConnectionError("connection closed"))
+        self._pending.clear()
+
+    async def _read_loop(self) -> None:
+        proc = self._proc
+        if proc is None or proc.stdout is None or proc.stderr is None:
+            return
+        stderr_task = asyncio.create_task(self._drain_stderr(proc.stderr))
+        try:
+            while True:
+                line = await proc.stdout.readline()
+                if not line:
+                    break
+                self._dispatch(line)
+        finally:
+            _ = stderr_task.cancel()
+            for future in self._pending.values():
+                if not future.done():
+                    future.set_exception(McpConnectionError("server closed the stream"))
+            self._pending.clear()
+
+    async def _drain_stderr(self, stderr: asyncio.StreamReader) -> None:
+        while True:
+            line = await stderr.readline()
+            if not line:
+                return
+            self._stderr_tail.append(line.decode(errors="replace").rstrip())
+
+    def _dispatch(self, line: bytes) -> None:
+        try:
+            payload = _json_dict(cast("object", json.loads(line)))
+        except json.JSONDecodeError:
+            return
+        message_id = payload.get("id")
+        if message_id is not None:
+            if isinstance(message_id, bool) or not isinstance(message_id, (int, str)):
+                return
+            request_id = int(message_id)
+            future = self._pending.pop(request_id, None)
+            if future is None or future.done():
+                return
+            error = payload.get("error")
+            if error is not None:
+                err = _json_dict(error)
+                detail = str(err.get("message", "unknown error")) if err else str(error)
+                future.set_exception(McpConnectionError(detail))
+            else:
+                future.set_result(_json_dict(payload.get("result")))
+            return
+        # Notifications carry no id; only tool-list changes matter to the host.
+        if payload.get("method") == "notifications/tools/list_changed":
+            self.tools_stale = True
+
+    def _send(self, payload: Json) -> None:
+        proc = self._proc
+        stdin = proc.stdin if proc is not None else None
+        if stdin is None:
+            msg = "server process is not running"
+            raise McpConnectionError(msg)
+        _ = stdin.write(json.dumps(payload, ensure_ascii=False).encode() + b"\n")
+
+    async def _request(self, method: str, params: Json | None = None) -> Json:
+        timeout = self._config.timeout or DEFAULT_MCP_TIMEOUT
+        future: asyncio.Future[Json] = asyncio.get_running_loop().create_future()
+        request_id = self._next_id
+        self._next_id += 1
+        self._pending[request_id] = future
+        payload: Json = {"jsonrpc": "2.0", "id": request_id, "method": method}
+        if params is not None:
+            payload["params"] = params
+        try:
+            self._send(payload)
+        except McpConnectionError:
+            _ = self._pending.pop(request_id, None)
+            raise
+        try:
+            return await asyncio.wait_for(future, timeout)
+        except TimeoutError:
+            _ = self._pending.pop(request_id, None)
+            msg = f"{method} timed out after {timeout}s"
+            raise McpConnectionError(msg) from None
+
+    async def _notify(self, method: str, params: Json | None = None) -> None:
+        payload: Json = {"jsonrpc": "2.0", "method": method}
+        if params is not None:
+            payload["params"] = params
+        self._send(payload)
+
+    async def _initialize(self) -> None:
+        _ = await self._request(
+            "initialize",
+            {
+                "protocolVersion": MCP_PROTOCOL_VERSION,
+                "capabilities": {},
+                "clientInfo": {"name": MCP_CLIENT_NAME, "version": _client_version()},
+            },
+        )
+        await self._notify("notifications/initialized")
+
+    async def list_tools(self) -> list[McpTool]:
+        """Fetch the full tool list, following ``nextCursor`` pagination."""
+        tools: list[McpTool] = []
+        cursor: str | None = None
+        for _ in range(_MAX_TOOL_PAGES):
+            params: Json = {} if cursor is None else {"cursor": cursor}
+            result = await self._request("tools/list", params)
+            tools.extend(McpTool.from_raw(item) for item in _json_list(result.get("tools")))
+            next_cursor = result.get("nextCursor")
+            if not isinstance(next_cursor, str) or not next_cursor:
+                return tools
+            cursor = next_cursor
+        return tools
+
+    async def call_tool(self, name: str, arguments: Json) -> str:
+        """Run ``tools/call`` and return the flattened text result.
+
+        Tool-level failures (``isError``) come back as ``error: …`` text, not
+        exceptions: the protocol round-trip itself succeeded.
+        """
+        result = await self._request("tools/call", {"name": name, "arguments": arguments})
+        text = _format_content(result)
+        if result.get("isError") is True:
+            return f"error: {text or 'tool reported failure without content'}"
+        return text
+
+
+class McpManager:
+    """All configured MCP server connections for this process."""
+
+    _config: McpConfig
+    _connections: dict[str, McpServerConnection]
+
+    def __init__(self, config: McpConfig) -> None:
+        self._config = config
+        self._connections = {}
+
+    def enabled_servers(self) -> dict[str, McpServerConfig]:
+        """Configured servers minus the ``disable`` list, in config order."""
+        disabled = {name.strip() for name in self._config.disable}
+        return {name: cfg for name, cfg in self._config.servers.items() if name not in disabled}
+
+    async def ensure_started(self) -> None:
+        """Connect every enabled server that has no live connection yet."""
+        for name, cfg in self.enabled_servers().items():
+            connection = self._connections.get(name)
+            if connection is not None and connection.connected:
+                continue
+            if connection is not None:
+                await connection.aclose()
+            connection = McpServerConnection(name, cfg)
+            self._connections[name] = connection
+            await connection.start()
+
+    async def restart(self, config: McpConfig) -> None:
+        """Drop every connection and adopt a new config (e.g. /mcp reconnect)."""
+        await self.aclose()
+        self._config = config
+
+    async def aclose(self) -> None:
+        for connection in self._connections.values():
+            await connection.aclose()
+        self._connections.clear()
+
+    def statuses(self) -> list[tuple[str, str, int]]:
+        """(server, status, tool_count) rows for diagnostics surfaces."""
+        rows: list[tuple[str, str, int]] = []
+        for name in self.enabled_servers():
+            connection = self._connections.get(name)
+            if connection is None:
+                rows.append((name, "not started", 0))
+            else:
+                rows.append((name, connection.status, len(connection.tools)))
+        return rows
+
+    def connections(self) -> list[McpServerConnection]:
+        return list(self._connections.values())
+
+    async def refresh_tools(self) -> None:
+        """Re-run tools/list on every connected server."""
+        for connection in self._connections.values():
+            if connection.connected:
+                connection.tools = await connection.list_tools()
+                connection.tools_stale = False
+
+    async def call(self, server: str, tool: str, arguments: Json) -> str:
+        connection = self._connections.get(server)
+        if connection is None or not connection.connected:
+            msg = f"server {server!r} is not connected"
+            raise McpConnectionError(msg)
+        return await connection.call_tool(tool, arguments)

src/plyngent/tools/catalog.py

@@ -12,7 +12,7 @@ from plyngent.agent.tools import ToolDefinition, ToolTag
 if TYPE_CHECKING:
     from collections.abc import Container, Generator
 
-type ToolSourceKind = Literal["builtin", "plugin"]
+type ToolSourceKind = Literal["builtin", "plugin", "mcp"]
 type ToolSurface = Literal["local", "public"]
 
 
@@ -29,8 +29,8 @@ class ToolSource:
     def __str__(self) -> str:
         if self.kind == "builtin":
             return "builtin"
-        plugin = self.plugin_id or "?"
-        return f"plugin:{plugin}"
+        id_part = self.plugin_id or "?"
+        return f"{self.kind}:{id_part}"
 
 
 @dataclass(frozen=True, slots=True)
@@ -152,6 +152,13 @@ class ToolCatalog:
     def clear(self) -> None:
         self._by_name.clear()
 
+    def purge_kind(self, kind: ToolSourceKind) -> int:
+        """Remove every entry registered from *kind*; return the count removed."""
+        doomed = [name for name, entry in self._by_name.items() if entry.source.kind == kind]
+        for name in doomed:
+            del self._by_name[name]
+        return len(doomed)
+
     def snapshot(self) -> dict[str, RegisteredTool]:
         return dict(self._by_name)
 

src/plyngent/tools/mcp.py

@@ -0,0 +1,108 @@
+"""Register MCP server tools into the tool catalog (namespaced, LOCAL tags).
+
+Mirrors :mod:`plyngent.tools.plugins`: the catalog only *registers* what the
+model may see; hosts still select into a :class:`ToolRegistry`. Connections
+must already be started (see :class:`~plyngent.runtime.mcp_client.McpManager`);
+this module is sync-only so registry rebuilds never need an event loop.
+
+Tool names are namespaced ``mcp__<server>__<tool>`` so two servers (or a
+server and a builtin) can never collide on sanitized names; the catalog still
+refuses true clashes.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+from plyngent.agent.tools import ToolDefinition, ToolTag
+from plyngent.tools.catalog import ToolSource, get_catalog
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+    from plyngent.config.models import McpConfig
+    from plyngent.runtime.mcp_client import McpManager, McpTool
+
+_MCP_PREFIX = "mcp__"
+_NAME_SEGMENT_MAX = 64
+_SANITIZE_PATTERN = re.compile(r"[^A-Za-z0-9_-]")
+
+
+def _sanitize_segment(part: str) -> str:
+    """Reduce a server/tool name to a model-safe, bounded name segment."""
+    return _SANITIZE_PATTERN.sub("_", part)[:_NAME_SEGMENT_MAX] or "srv"
+
+
+def mcp_tool_name(server: str, tool: str) -> str:
+    """Namespaced catalog name for one MCP tool (collision-free per server)."""
+    return f"{_MCP_PREFIX}{_sanitize_segment(server)}__{_sanitize_segment(tool)}"
+
+
+def _make_handler(
+    manager: McpManager,
+    server: str,
+    tool: str,
+) -> Callable[..., Awaitable[str]]:
+    async def handler(**arguments: object) -> str:
+        return await manager.call(server, tool, dict(arguments))
+
+    return handler
+
+
+def register_mcp_server_tools(
+    server: str,
+    tools: list[McpTool],
+    *,
+    manager: McpManager,
+    read_only: bool = False,
+) -> list[str]:
+    """Register one server's advertised tools; returns the registered names.
+
+    Tags: LOCAL (plus READ_ONLY when the server config opts in) — never YOLO or
+    TRUSTABLE, so host-side danger classification always sees every call.
+    """
+    catalog = get_catalog()
+    source = ToolSource(kind="mcp", plugin_id=server)
+    names: list[str] = []
+    for entry in tools:
+        if not entry.name:
+            continue
+        tags = ToolTag.LOCAL
+        if read_only:
+            tags |= ToolTag.READ_ONLY
+        description = entry.description or f"MCP tool {entry.name!r} from server {server!r}"
+        definition = ToolDefinition(
+            name=mcp_tool_name(server, entry.name),
+            description=f"[mcp:{server}] {description}",
+            parameters=dict(entry.input_schema),
+            handler=_make_handler(manager, server, entry.name),
+            tags=tags,
+        )
+        catalog.register(definition, source=source)
+        names.append(definition.name)
+    return names
+
+
+def purge_mcp_tools() -> int:
+    """Remove previously registered MCP tools (call before re-registering)."""
+    return get_catalog().purge_kind("mcp")
+
+
+def register_mcp_tools(manager: McpManager, config: McpConfig) -> list[str]:
+    """Register every connected server's tools; replaces prior MCP entries."""
+    _ = purge_mcp_tools()
+    names: list[str] = []
+    for connection in manager.connections():
+        if not connection.connected or not connection.tools:
+            continue
+        server_cfg = config.servers.get(connection.name)
+        names.extend(
+            register_mcp_server_tools(
+                connection.name,
+                connection.tools,
+                manager=manager,
+                read_only=server_cfg.read_only if server_cfg is not None else False,
+            )
+        )
+    return names

tests/test_cli/test_mcp_cmd.py

@@ -0,0 +1,122 @@
+"""CLI slash ``/mcp`` command and MCP wiring in ``ReplState``."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import pytest
+import tomlkit
+
+from plyngent.cli.slash import handle_slash
+from plyngent.cli.state import ReplState
+from plyngent.config.models import DatabaseConfig, McpConfig, McpServerConfig, OpenAIProvider
+from plyngent.config.store import ConfigStore
+from plyngent.memory import MemoryStore
+from plyngent.runtime.mcp_client import McpManager
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+_FAKE_SERVER = Path(__file__).parent.parent / "test_runtime" / "mcp_fake_server.py"
+
+
+def _server_config(**overrides: object) -> McpServerConfig:
+    base: dict[str, object] = {
+        "command": sys.executable,
+        "args": [str(_FAKE_SERVER)],
+        "timeout": 5.0,
+    }
+    return McpServerConfig(**{**base, **overrides})  # type: ignore[arg-type]
+
+
+@pytest.fixture
+async def state(tmp_path: Path) -> AsyncIterator[ReplState]:
+    memory = await MemoryStore.open(DatabaseConfig())
+    provider = OpenAIProvider(access_key_or_token="sk-test")
+    config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
+    config.providers = {"local": provider}
+    st = ReplState(
+        config=config,
+        memory=memory,
+        workspace=tmp_path,
+        provider_name="local",
+        provider=provider,
+        model="gpt-test",
+        tools_enabled=False,
+    )
+    await st.new_session("t")
+    yield st
+    await memory.close()
+
+
+async def test_slash_mcp_list_without_manager(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
+    assert await handle_slash(state, "/mcp") is True
+    assert "no servers configured" in capsys.readouterr().out
+    assert await handle_slash(state, "/mcp list") is True
+    assert "no servers configured" in capsys.readouterr().out
+
+
+async def test_slash_mcp_list_statuses(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
+    manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
+    await manager.ensure_started()
+    state.mcp_manager = manager
+    try:
+        assert await handle_slash(state, "/mcp list") is True
+        out = capsys.readouterr().out
+        assert "fake" in out
+        assert "connected" in out
+        assert "tools=3" in out
+    finally:
+        await manager.aclose()
+
+
+async def test_slash_mcp_reconnect_connects_new_config(
+    state: ReplState,
+    capsys: pytest.CaptureFixture[str],
+    tmp_path: Path,
+) -> None:
+    # The fixture's config file does not exist yet; write one with an MCP server.
+    payload = (
+        "[mcp.servers.fake]\n"
+        f"command = {json.dumps(sys.executable)}\n"
+        f"args = [{json.dumps(str(_FAKE_SERVER))}]\n"
+        "timeout = 5.0\n"
+    )
+    _ = (tmp_path / "plyngent.toml").write_text(payload, encoding="utf-8")
+    assert await handle_slash(state, "/mcp reconnect") is True
+    try:
+        assert state.mcp_manager is not None
+        statuses = state.mcp_manager.statuses()
+        assert [name for name, _, _ in statuses] == ["fake"]
+        assert statuses[0][1] == "connected"
+        assert statuses[0][2] == 3
+        out = capsys.readouterr().out
+        assert "tools=off" in out
+        assert "fake" in out
+    finally:
+        if state.mcp_manager is not None:
+            await state.mcp_manager.aclose()
+
+
+async def test_mcp_tools_registered_on_tools_rebuild(state: ReplState) -> None:
+    manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
+    await manager.ensure_started()
+    state.mcp_manager = manager
+    try:
+        from plyngent.tools.catalog import catalog_scope
+
+        state.tools_enabled = True
+        with catalog_scope(empty=True):
+            state.rebuild_client()
+        registry = state.agent.tools
+        assert registry is not None
+        assert registry.get("mcp__fake__echo") is not None
+        assert registry.get("mcp__fake__fail") is not None
+        # Handler proxies to the live server subprocess.
+        out = await registry.execute("mcp__fake__echo", '{"text": "hi"}')
+        assert out == "echo: hi"
+    finally:
+        await manager.aclose()

tests/test_config/test_mcp_section.py

@@ -0,0 +1,66 @@
+"""[mcp] section parse: server definitions, disable gate, malformed fallback."""
+
+from __future__ import annotations
+
+from plyngent.config import load
+
+
+def test_mcp_defaults_and_parse(tmp_path) -> None:
+    path = tmp_path / "c.toml"
+    _ = path.write_text("", encoding="utf-8")
+    store = load(path)
+    assert store.mcp_config.servers == {}
+    assert store.mcp_config.disable == []
+
+    path.write_text(
+        """
+[mcp.servers.docs]
+command = "npx"
+args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
+env = { RUST_LOG = "info" }
+timeout = 10.0
+read_only = true
+
+[mcp.servers.search]
+command = "uvx"
+args = ["mcp-server-fetch"]
+
+[mcp]
+disable = ["search"]
+""",
+        encoding="utf-8",
+    )
+    store = load(path)
+    mcp = store.mcp_config
+    assert set(mcp.servers) == {"docs", "search"}
+    docs = mcp.servers["docs"]
+    assert docs.command == "npx"
+    assert docs.args == ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
+    assert docs.env == {"RUST_LOG": "info"}
+    assert docs.timeout == 10.0
+    assert docs.read_only is True
+    assert docs.url == ""
+    assert mcp.disable == ["search"]
+
+
+def test_mcp_malformed_section_falls_back_to_defaults(tmp_path) -> None:
+    path = tmp_path / "c.toml"
+    # Unknown section key.
+    path.write_text("[mcp]\nenabled = true\n", encoding="utf-8")
+    store = load(path)
+    assert store.mcp_config.servers == {}
+
+    # Unknown server key (typo).
+    path.write_text('[mcp.servers.docs]\ncmds = "npx"\n', encoding="utf-8")
+    store = load(path)
+    assert store.mcp_config.servers == {}
+
+    # Server entry is not a table.
+    path.write_text("[mcp]\nservers = 3\n", encoding="utf-8")
+    store = load(path)
+    assert store.mcp_config.servers == {}
+
+    # Empty server name.
+    path.write_text('[mcp.servers.""]\ncommand = "npx"\n', encoding="utf-8")
+    store = load(path)
+    assert store.mcp_config.servers == {}

tests/test_runtime/mcp_fake_server.py

@@ -0,0 +1,96 @@
+"""Minimal MCP stdio server used by the client tests (echo + sleep tools)."""
+
+from __future__ import annotations
+
+import json
+import sys
+
+
+def respond(request_id: int, result: object) -> None:
+    sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result}) + "\n")
+    sys.stdout.flush()
+
+
+def main() -> None:
+    for line in sys.stdin:
+        try:
+            payload = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        method = payload.get("method", "")
+        request_id = payload.get("id")
+        if request_id is None:
+            continue  # notification (notifications/initialized)
+        if method == "initialize":
+            respond(
+                int(request_id),
+                {
+                    "protocolVersion": "2025-06-18",
+                    "capabilities": {"tools": {}},
+                    "serverInfo": {"name": "fake-mcp", "version": "0.1.0"},
+                },
+            )
+        elif method == "tools/list":
+            respond(
+                int(request_id),
+                {
+                    "tools": [
+                        {
+                            "name": "echo",
+                            "description": "Echo the given text back.",
+                            "inputSchema": {
+                                "type": "object",
+                                "properties": {"text": {"type": "string"}},
+                                "required": ["text"],
+                            },
+                        },
+                        {
+                            "name": "fail",
+                            "description": "Always reports a tool error.",
+                            "inputSchema": {"type": "object", "properties": {}},
+                        },
+                        {
+                            "name": "slow",
+                            "description": "Sleeps past any client timeout.",
+                            "inputSchema": {"type": "object", "properties": {}},
+                        },
+                    ]
+                },
+            )
+        elif method == "tools/call":
+            name = (payload.get("params") or {}).get("name")
+            if name == "echo":
+                text = str((payload.get("params") or {}).get("arguments", {}).get("text", ""))
+                respond(
+                    int(request_id),
+                    {
+                        "content": [{"type": "text", "text": f"echo: {text}"}],
+                        "isError": False,
+                    },
+                )
+            elif name == "fail":
+                respond(
+                    int(request_id),
+                    {"content": [{"type": "text", "text": "boom"}], "isError": True},
+                )
+            else:
+                import time
+
+                time.sleep(5)
+                respond(int(request_id), {"content": [], "isError": False})
+        else:
+            sys.stdout.write(
+                json.dumps(
+                    {
+                        "jsonrpc": "2.0",
+                        "id": request_id,
+                        "error": {"code": -32601, "message": f"unknown method {method}"},
+                    }
+                )
+                + "\n"
+            )
+            sys.stdout.flush()
+
+
+if __name__ == "__main__":
+    main()

tests/test_runtime/test_mcp_client.py

@@ -0,0 +1,106 @@
+"""MCP stdio client + manager against the bundled fake server subprocess."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from plyngent.config.models import McpConfig, McpServerConfig
+from plyngent.runtime.mcp_client import (
+    McpConnectionError,
+    McpManager,
+    McpServerConnection,
+)
+
+_FAKE_SERVER = Path(__file__).parent / "mcp_fake_server.py"
+
+
+def _server_config(**overrides: object) -> McpServerConfig:
+    base: dict[str, object] = {
+        "command": sys.executable,
+        "args": [str(_FAKE_SERVER)],
+        "timeout": 5.0,
+    }
+    return McpServerConfig(**{**base, **overrides})  # type: ignore[arg-type]
+
+
+async def test_connection_handshake_lists_and_calls_tools() -> None:
+    connection = McpServerConnection("fake", _server_config())
+    await connection.start()
+    try:
+        assert connection.connected
+        assert connection.error is None
+        assert [tool.name for tool in connection.tools] == ["echo", "fail", "slow"]
+        echo = connection.tools[0]
+        assert echo.description == "Echo the given text back."
+        assert echo.input_schema["type"] == "object"
+
+        result = await connection.call_tool("echo", {"text": "hello"})
+        assert result == "echo: hello"
+    finally:
+        await connection.aclose()
+
+
+async def test_tool_error_returns_error_text() -> None:
+    connection = McpServerConnection("fake", _server_config())
+    await connection.start()
+    try:
+        result = await connection.call_tool("fail", {})
+        assert result == "error: boom"
+    finally:
+        await connection.aclose()
+
+
+async def test_request_timeout_raises() -> None:
+    connection = McpServerConnection("fake", _server_config(timeout=0.2))
+    await connection.start()
+    try:
+        with pytest.raises(McpConnectionError, match="timed out"):
+            await connection.call_tool("slow", {})
+    finally:
+        await connection.aclose()
+
+
+async def test_spawn_failure_records_error() -> None:
+    connection = McpServerConnection("bad", _server_config(command="/nonexistent/plyngent-mcp-binary"))
+    await connection.start()
+    assert not connection.connected
+    assert connection.error is not None
+    assert "spawn" in connection.status
+    await connection.aclose()
+
+
+async def test_close_terminates_subprocess() -> None:
+    connection = McpServerConnection("fake", _server_config())
+    await connection.start()
+    proc = connection._proc
+    assert proc is not None
+    await connection.aclose()
+    assert connection._proc is None
+    assert proc.returncode is not None
+
+
+async def test_manager_start_status_and_call() -> None:
+    manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=[]))
+    await manager.ensure_started()
+    try:
+        assert [(name, status, count) for name, status, count in manager.statuses()] == [("fake", "connected", 3)]
+        assert await manager.call("fake", "echo", {"text": "hi"}) == "echo: hi"
+    finally:
+        await manager.aclose()
+    assert manager.statuses() == [("fake", "not started", 0)]
+
+
+async def test_manager_disable_and_restart() -> None:
+    manager = McpManager(McpConfig(servers={"fake": _server_config()}, disable=["fake"]))
+    await manager.ensure_started()
+    assert manager.connections() == []
+    with pytest.raises(McpConnectionError, match="not connected"):
+        await manager.call("fake", "echo", {})
+
+    await manager.restart(McpConfig(servers={"fake": _server_config()}, disable=[]))
+    await manager.ensure_started()
+    assert manager.statuses() == [("fake", "connected", 3)]
+    await manager.aclose()

tests/test_tools/test_mcp_tools.py

@@ -0,0 +1,93 @@
+"""MCP tool registration: namespacing, tags, catalog source, purge, handlers."""
+
+from __future__ import annotations
+
+import pytest
+
+from plyngent.agent.tools import ToolTag
+from plyngent.runtime.mcp_client import McpTool
+from plyngent.tools.catalog import catalog_scope
+from plyngent.tools.mcp import mcp_tool_name, purge_mcp_tools, register_mcp_server_tools
+
+
+class _StubManager:
+    """Duck-typed McpManager stand-in (records calls, returns canned text)."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str, dict[str, object]]] = []
+
+    async def call(self, server: str, tool: str, arguments: dict[str, object]) -> str:
+        self.calls.append((server, tool, arguments))
+        return f"{server}:{tool}:{arguments.get('text', '')}"
+
+
+def _tools() -> list[McpTool]:
+    return [
+        McpTool(
+            name="echo",
+            description="Echo text.",
+            input_schema={"type": "object", "properties": {"text": {"type": "string"}}},
+        ),
+        McpTool(name="", description="skipped: empty name", input_schema={}),
+    ]
+
+
+def test_mcp_tool_name_sanitizes_and_namespaces() -> None:
+    assert mcp_tool_name("my server", "a/b c") == "mcp__my_server__a_b_c"
+    assert mcp_tool_name("x", "y").startswith("mcp__")
+
+
+def test_register_names_tags_source_and_purge() -> None:
+    manager = _StubManager()
+    with catalog_scope(empty=True):
+        names = register_mcp_server_tools("docs", _tools(), manager=manager)  # type: ignore[arg-type]
+        assert names == ["mcp__docs__echo"]
+
+        from plyngent.tools.catalog import get_catalog
+
+        entry = get_catalog().get("mcp__docs__echo")
+        assert entry is not None
+        assert entry.source.kind == "mcp"
+        assert entry.source.plugin_id == "docs"
+        assert str(entry.source) == "mcp:docs"
+        definition = entry.definition
+        assert definition.tags == ToolTag.LOCAL
+        assert definition.description.startswith("[mcp:docs]")
+        assert definition.parameters["type"] == "object"
+        assert "text" in definition.parameters["properties"]
+
+        assert purge_mcp_tools() == 1
+        assert get_catalog().get("mcp__docs__echo") is None
+        assert purge_mcp_tools() == 0
+
+
+def test_register_read_only_flag() -> None:
+    manager = _StubManager()
+    with catalog_scope(empty=True):
+        register_mcp_server_tools("docs", _tools()[:1], manager=manager, read_only=True)  # type: ignore[arg-type]
+        from plyngent.tools.catalog import get_catalog
+
+        entry = get_catalog().get("mcp__docs__echo")
+        assert entry is not None
+        assert entry.definition.tags == ToolTag.LOCAL | ToolTag.READ_ONLY
+
+
+def test_duplicate_registration_collides() -> None:
+    manager = _StubManager()
+    with catalog_scope(empty=True):
+        register_mcp_server_tools("docs", _tools()[:1], manager=manager)  # type: ignore[arg-type]
+        with pytest.raises(ValueError, match="collision"):
+            register_mcp_server_tools("docs", _tools()[:1], manager=manager)  # type: ignore[arg-type]
+
+
+async def test_handler_proxies_to_manager_call() -> None:
+    manager = _StubManager()
+    with catalog_scope(empty=True):
+        register_mcp_server_tools("docs", _tools()[:1], manager=manager)  # type: ignore[arg-type]
+        from plyngent.tools.catalog import get_catalog
+
+        entry = get_catalog().get("mcp__docs__echo")
+        assert entry is not None
+        result = await entry.definition.handler(text="hi")  # type: ignore[union-attr]
+        assert result == "docs:echo:hi"
+        assert manager.calls == [("docs", "echo", {"text": "hi"})]