feat: implement mcp frontend

src/beaver_gateway/mcp/client_pool.py

@@ -16,8 +16,11 @@ from fastmcp import Client
 from fastmcp.client.transports import StdioTransport, StreamableHttpTransport
 from fastmcp.server import create_proxy
 
+from beaver_gateway.mcp.lenient import LenientStdioTransport
+
 if TYPE_CHECKING:
     from fastmcp import FastMCP
+    from fastmcp.client.transports.base import ClientTransport
 
     from beaver_gateway.mcp.types import HttpMcp, StdioMcp
 
@@ -27,22 +30,37 @@ def build_stdio_proxy(spec: StdioMcp) -> FastMCP:
 
     ``spec.command`` is a non-empty tuple; the first element is the
     executable and the rest are CLI args. ``StdioTransport`` keeps the
-    subprocess alive across calls.
+    subprocess alive across calls. When ``spec.lenient`` is set we
+    swap in :class:`LenientStdioTransport`, which silently drops
+    non-JSON-RPC stdout lines instead of forwarding them as exceptions
+    (lets us ingest MCPs that print ``Processing...`` chatter on stdout
+    without leaking warnings to downstream UIs).
     """
     if not spec.command:
         msg = f"stdio MCP {spec.name!r} has empty command"
         raise ValueError(msg)
     command, *args = spec.command
-    transport = StdioTransport(
-        command=command,
-        args=list(args),
-        env=spec.env,
-        cwd=str(spec.cwd) if spec.cwd is not None else None,
-    )
+    cwd = str(spec.cwd) if spec.cwd is not None else None
+    transport: ClientTransport
+    if spec.lenient:
+        transport = LenientStdioTransport(
+            command=command, args=list(args), env=spec.env, cwd=cwd
+        )
+    else:
+        transport = StdioTransport(
+            command=command, args=list(args), env=spec.env, cwd=cwd
+        )
     return create_proxy(Client(transport, name=spec.name))
 
 
 def build_http_proxy(spec: HttpMcp) -> FastMCP:
-    """Wrap a remote streamable-HTTP MCP into a mountable ``FastMCPProxy``."""
-    transport = StreamableHttpTransport(url=spec.url, auth=spec.auth)
+    """Wrap a remote streamable-HTTP MCP into a mountable ``FastMCPProxy``.
+
+    ``spec.headers`` are folded into the upstream client; pass per-call
+    auth (e.g. ``X-Api-Key``) here rather than as ``auth`` when the
+    upstream rejects Bearer headers.
+    """
+    transport = StreamableHttpTransport(
+        url=spec.url, auth=spec.auth, headers=spec.headers
+    )
     return create_proxy(Client(transport, name=spec.name))

src/beaver_gateway/mcp/internal_app.py

@@ -1,4 +1,4 @@
-"""Internal MCP aggregator — one ASGI app, N FastMCP namespaces.
+"""Internal MCP aggregator — one ASGI app, N FastMCP namespaces + ``all``.
 
 Each ``McpServer`` declared in the user's config becomes its own
 ``FastMCP`` instance (regular for ``python_tool``, ``FastMCPProxy`` for
@@ -9,9 +9,18 @@ namespace via loopback as a distinct MCP server URL — preserving
 per-domain framing while costing only one process worth of RAM
 (PRD §6).
 
+Phase 3 adds ``/mcp/all/``: a single FastMCP whose tools are the union
+of every namespace's tools, prefixed by FastMCP's ``namespace_<tool>``
+convention (e.g. ``time_current_time``). It's the escape-hatch for
+clients that can only configure one MCP server — discouraged for tool-
+heavy setups (PRD §6 cites the ~95%→~71% tool-selection drop on flat
+namespaces) but real and reachable.
+
 The aggregator returns both the app and a ``{name: url}`` map; Phase
 2.2's ``ClaudeCodeBackendAdapter`` plugs the map directly into
-``BackendOptions.mcp_servers``.
+``BackendOptions.mcp_servers``. ``/mcp/all/`` is NOT included in that
+map — claude-code-agents always get per-domain framing; only the
+external MCP frontend (Phase 3.1) reverse-proxies the flat endpoint.
 """
 
 from __future__ import annotations
@@ -19,6 +28,7 @@ from __future__ import annotations
 from contextlib import AsyncExitStack, asynccontextmanager
 from typing import TYPE_CHECKING
 
+from fastmcp import FastMCP
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -29,30 +39,48 @@ from beaver_gateway.mcp.wrap import build_python_tool_server
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator, Iterable
 
-    from fastmcp import FastMCP
-
     from beaver_gateway.mcp.types import McpServerT
 
 
+ALL_NAMESPACE = "all"
+"""URL segment for the flat-namespace aggregator (``/mcp/all/``)."""
+
+
 def build_internal_app(
     mcps: Iterable[McpServerT], *, host: str, port: int
 ) -> tuple[Starlette, dict[str, str]]:
-    """Build the aggregator ``Starlette`` app and the ``{name: url}`` map.
+    """Build the aggregator ``Starlette`` app and the per-namespace URL map.
 
     ``host``/``port`` only flavour the URL strings handed back — actually
     listening on them is the caller's job (``cli.main`` runs a uvicorn
     server in a TaskGroup). We accept the address here so callers don't
     have to format the URLs themselves and risk drifting from the
     ``/mcp/<name>`` convention.
+
+    Returns a map of ``{namespace: url}`` for the per-domain endpoints
+    only (claude-code's MCP routing expects per-domain framing). The
+    ``/mcp/all/`` bundle endpoint exists on the same app but is
+    intentionally omitted from the URL map — it's only meaningful to
+    external clients via the MCP frontend, not to claude-code.
     """
     servers: dict[str, FastMCP] = {spec.name: _build_server(spec) for spec in mcps}
 
-    child_apps = [s.http_app(transport="http", path="/") for s in servers.values()]
+    child_apps = {
+        name: s.http_app(transport="http", path="/")
+        for name, s in servers.items()
+    }
     routes = [
-        Mount(f"/mcp/{name}", app=app)
-        for name, app in zip(servers, child_apps, strict=True)
+        Mount(f"/mcp/{name}", app=app) for name, app in child_apps.items()
     ]
 
+    # /mcp/all — flat-namespace bundle. Skip when there's nothing to
+    # bundle so we don't pay for an empty session manager lifecycle.
+    all_app = None
+    if servers:
+        all_server = _build_all_server(servers)
+        all_app = all_server.http_app(transport="http", path="/")
+        routes.append(Mount(f"/mcp/{ALL_NAMESPACE}", app=all_app))
+
     @asynccontextmanager
     async def lifespan(_parent: Starlette) -> AsyncIterator[None]:
         # Each FastMCP http_app stores its session manager init in its
@@ -61,8 +89,12 @@ def build_internal_app(
         # children come up together and unwind in reverse order on
         # shutdown.
         async with AsyncExitStack() as stack:
-            for child in child_apps:
+            for child in child_apps.values():
                 await stack.enter_async_context(child.router.lifespan_context(child))
+            if all_app is not None:
+                await stack.enter_async_context(
+                    all_app.router.lifespan_context(all_app)
+                )
             yield
 
     app = Starlette(routes=routes, lifespan=lifespan)
@@ -85,3 +117,22 @@ def _build_server(spec: McpServerT) -> FastMCP:
     # type-narrowing honest if a new variant lands without updates here.
     msg = f"unsupported McpServer variant: {type(spec).__name__}"
     raise TypeError(msg)
+
+
+def _build_all_server(children: dict[str, FastMCP]) -> FastMCP:
+    """Compose a single FastMCP whose tools union every namespace's tools.
+
+    FastMCP's ``mount(namespace=...)`` namespaces every tool as
+    ``<namespace>_<original_name>``, so the flat endpoint becomes
+    ``time_current_time``, ``calendar_event_create``, etc.
+
+    Mounted children are accessed in-memory via ``FastMCPProvider`` —
+    requests don't bounce through the child's own ``http_app``, so this
+    aggregator has its own independent streamable-HTTP session manager
+    and lifespan, and the per-namespace ``/mcp/<name>/`` mounts keep
+    working unchanged.
+    """
+    parent = FastMCP(name="beaver-gateway-all")
+    for name, child in children.items():
+        parent.mount(child, namespace=name)
+    return parent

src/beaver_gateway/mcp/lenient.py

@@ -0,0 +1,234 @@
+"""Tolerant transports for upstream MCPs that don't strictly speak JSON-RPC.
+
+Some real-world MCP servers print non-JSON chatter to stdout before / between
+their actual JSON-RPC frames (``Processing...``, banners, dependency-load
+messages, etc.). The reference ``mcp.client.stdio.stdio_client`` parses every
+stdout line as JSON-RPC and ships any parse failure downstream as an
+exception, which the MCP ``ClientSession`` then logs as a warning that bleeds
+into client UIs (Cursor, Cline) when they connect through us.
+
+``LenientStdioTransport`` re-implements the stdio-client wiring with one
+behavioural change: lines that don't parse as JSON-RPC are *silently
+dropped* (one ``DEBUG`` log entry, no exception forwarded). Downstream
+consumers see only valid messages. We keep the rest of the contract identical
+to the reference client, including the spec-mandated graceful shutdown
+sequence (close stdin → wait → SIGTERM → SIGKILL).
+
+The transport plugs into ``fastmcp.server.create_proxy`` just like
+``StdioTransport`` does, so the rest of the aggregator doesn't need to
+know which flavour it got.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import sys
+from typing import TYPE_CHECKING
+
+import anyio
+import anyio.lowlevel
+from anyio.streams.text import TextReceiveStream
+from fastmcp.client.transports.base import ClientTransport
+from mcp import ClientSession, types
+from mcp.client.stdio import (
+    PROCESS_TERMINATION_TIMEOUT,
+    StdioServerParameters,
+    _create_platform_compatible_process,
+    _get_executable_command,
+    _terminate_process_tree,
+)
+from mcp.shared.message import SessionMessage
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+    from typing import TextIO, Unpack
+
+    from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+    from fastmcp.client.transports.base import SessionKwargs
+
+
+_log = logging.getLogger("beaver_gateway.mcp.lenient")
+
+
+class LenientStdioTransport(ClientTransport):
+    """Stdio transport that tolerates non-JSON-RPC stdout noise.
+
+    Behaves like :class:`fastmcp.client.transports.StdioTransport` from the
+    consumer's perspective: one ``ClientSession`` per ``connect_session``
+    block, subprocess scoped to that block. We don't replicate the upstream
+    ``keep_alive`` flag because the only caller (``create_proxy``) opens
+    the session lazily on first request and keeps it open for the lifetime
+    of the proxy.
+    """
+
+    def __init__(
+        self,
+        command: str,
+        args: list[str],
+        env: dict[str, str] | None = None,
+        cwd: str | None = None,
+        log_file: TextIO | None = None,
+    ) -> None:
+        self.command = command
+        self.args = args
+        self.env = env
+        self.cwd = cwd
+        # TextIO only — pre-open Path callers themselves. The upstream
+        # ``StdioTransport`` opens Path for you, but we keep this thin so
+        # the type contract stays narrow and easy to validate.
+        self.log_file = log_file
+
+    @contextlib.asynccontextmanager
+    async def connect_session(
+        self, **session_kwargs: Unpack[SessionKwargs]
+    ) -> AsyncIterator[ClientSession]:
+        errlog: TextIO = self.log_file if self.log_file is not None else sys.stderr
+        async with _lenient_stdio_client(
+            StdioServerParameters(
+                command=self.command,
+                args=self.args,
+                env=self.env,
+                cwd=self.cwd,
+            ),
+            errlog=errlog,
+        ) as (read_stream, write_stream), ClientSession(
+            read_stream, write_stream, **session_kwargs
+        ) as session:
+            yield session
+
+    def __repr__(self) -> str:
+        return (
+            f"<LenientStdioTransport(command={self.command!r}, args={self.args!r})>"
+        )
+
+
+@contextlib.asynccontextmanager
+async def _lenient_stdio_client(  # noqa: PLR0915 — mirrors mcp.client.stdio.stdio_client
+    server: StdioServerParameters, errlog: TextIO = sys.stderr
+) -> AsyncIterator[
+    tuple[
+        MemoryObjectReceiveStream[SessionMessage | Exception],
+        MemoryObjectSendStream[SessionMessage],
+    ]
+]:
+    """Drop-in for ``mcp.client.stdio.stdio_client`` with a tolerant reader.
+
+    All differences from upstream live in ``stdout_reader``: lines that fail
+    ``JSONRPCMessage.model_validate_json`` are logged at DEBUG and skipped,
+    never forwarded as exceptions. This is what makes warning-noisy MCPs
+    quiet from the consumer's point of view.
+    """
+    read_stream_writer, read_stream = anyio.create_memory_object_stream[
+        SessionMessage | Exception
+    ](0)
+    write_stream, write_stream_reader = anyio.create_memory_object_stream[
+        SessionMessage
+    ](0)
+
+    try:
+        command = _get_executable_command(server.command)
+        env_default = _default_inherited_env()
+        process = await _create_platform_compatible_process(
+            command=command,
+            args=server.args,
+            env=(
+                {**env_default, **server.env}
+                if server.env is not None
+                else env_default
+            ),
+            errlog=errlog,
+            cwd=server.cwd,
+        )
+    except OSError:
+        await read_stream.aclose()
+        await write_stream.aclose()
+        await read_stream_writer.aclose()
+        await write_stream_reader.aclose()
+        raise
+
+    async def stdout_reader() -> None:
+        assert process.stdout, "Opened process is missing stdout"  # noqa: S101
+        try:
+            async with read_stream_writer:
+                buffer = ""
+                async for chunk in TextReceiveStream(
+                    process.stdout,
+                    encoding=server.encoding,
+                    errors=server.encoding_error_handler,
+                ):
+                    lines = (buffer + chunk).split("\n")
+                    buffer = lines.pop()
+                    for line in lines:
+                        stripped = line.strip()
+                        if not stripped:
+                            continue
+                        try:
+                            message = types.JSONRPCMessage.model_validate_json(
+                                stripped
+                            )
+                        except Exception:  # noqa: BLE001 — by design, see module doc
+                            _log.debug(
+                                "lenient stdio: dropped non-JSON line: %r",
+                                stripped[:200],
+                            )
+                            continue
+                        await read_stream_writer.send(SessionMessage(message))
+        except anyio.ClosedResourceError:
+            await anyio.lowlevel.checkpoint()
+
+    async def stdin_writer() -> None:
+        assert process.stdin, "Opened process is missing stdin"  # noqa: S101
+        try:
+            async with write_stream_reader:
+                async for session_message in write_stream_reader:
+                    payload = session_message.message.model_dump_json(
+                        by_alias=True, exclude_none=True
+                    )
+                    await process.stdin.send(
+                        (payload + "\n").encode(
+                            encoding=server.encoding,
+                            errors=server.encoding_error_handler,
+                        )
+                    )
+        except anyio.ClosedResourceError:
+            await anyio.lowlevel.checkpoint()
+
+    async with (
+        anyio.create_task_group() as tg,
+        process,
+    ):
+        tg.start_soon(stdout_reader)
+        tg.start_soon(stdin_writer)
+        try:
+            yield read_stream, write_stream
+        finally:
+            # MCP spec stdio shutdown: close stdin → wait → SIGTERM → SIGKILL.
+            if process.stdin:
+                with contextlib.suppress(Exception):
+                    await process.stdin.aclose()
+            try:
+                with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):
+                    await process.wait()
+            except TimeoutError:
+                await _terminate_process_tree(process)
+            except ProcessLookupError:
+                pass
+            await read_stream.aclose()
+            await write_stream.aclose()
+            await read_stream_writer.aclose()
+            await write_stream_reader.aclose()
+
+
+def _default_inherited_env() -> dict[str, str]:
+    """Same env shortlist as ``mcp.client.stdio.get_default_environment``.
+
+    Re-exported so we can compose with the user's ``env`` overrides without
+    forcing a private import in this module's body.
+    """
+    from mcp.client.stdio import get_default_environment
+
+    return get_default_environment()
+
+
+__all__ = ["LenientStdioTransport"]

src/beaver_gateway/mcp/types.py

@@ -23,20 +23,36 @@ class _BaseMcp(BaseModel):
 
 
 class StdioMcp(_BaseMcp):
-    """Subprocess MCP server we spawn and connect to over stdio."""
+    """Subprocess MCP server we spawn and connect to over stdio.
+
+    ``lenient`` switches the upstream stdio reader to a tolerant variant
+    that silently drops non-JSON-RPC lines from the subprocess's stdout
+    (``Processing...``-style chatter, banners, dependency-load messages).
+    The reference ``mcp.client.stdio.stdio_client`` forwards those parse
+    failures as exceptions, which bleed into Cursor/Cline UIs as warnings
+    when they connect through us. Default ``False`` keeps the strict
+    contract — flip it on per-namespace for known-noisy upstreams.
+    """
 
     kind: Literal["stdio"] = "stdio"
     command: tuple[str, ...]
     env: dict[str, str] | None = None
     cwd: Path | None = None
+    lenient: bool = False
 
 
 class HttpMcp(_BaseMcp):
-    """Remote MCP server reached over streamable HTTP."""
+    """Remote MCP server reached over streamable HTTP.
+
+    ``headers`` are forwarded on every request to the upstream MCP — handy
+    for upstreams that authenticate via custom header rather than a Bearer
+    token (``auth``).
+    """
 
     kind: Literal["http"] = "http"
     url: str
     auth: str | None = None
+    headers: dict[str, str] | None = None
 
 
 class PythonToolMcp(_BaseMcp):
@@ -62,19 +78,26 @@ class McpServer:
         command: Iterable[str],
         env: dict[str, str] | None = None,
         cwd: Path | str | None = None,
+        lenient: bool = False,
     ) -> StdioMcp:
         return StdioMcp(
             name=name,
             command=tuple(command),
             env=env,
             cwd=Path(cwd) if isinstance(cwd, str) else cwd,
+            lenient=lenient,
         )
 
     @classmethod
     def http(
-        cls, *, name: str, url: str, auth: str | None = None
+        cls,
+        *,
+        name: str,
+        url: str,
+        auth: str | None = None,
+        headers: dict[str, str] | None = None,
     ) -> HttpMcp:
-        return HttpMcp(name=name, url=url, auth=auth)
+        return HttpMcp(name=name, url=url, auth=auth, headers=headers)
 
     @classmethod
     def python_tool(