feat(mcp): discover and register tools on demand

packages/bub-acp-server/src/bub_acp_server/agent.py

@@ -962,7 +962,7 @@ async def _run_streaming_subagent(
         )
         child_state = {**context.state, "session_id": child_session}
         allowed_tools = resolve_tool_names(
-            param.allowed_tools or None, exclude={"subagent"}, all_names=agent.tools
+            param.allowed_tools or None, exclude={"subagent"}, all_names=agent.known_tools
         )
         await emitter.started(
             description=_subagent_description(param.prompt), model_id=param.model

packages/bub-acp-server/tests/test_mcp.py

@@ -112,7 +112,7 @@ async def call_echo(agent, session_id):
 async def runtime_tools(agent, session_id):
     inbound = agent._build_inbound([], agent._sessions[session_id])
     state = await agent.framework.build_state(inbound, inbound.session_id)
-    return state["_runtime_agent"].tools
+    return state["_runtime_agent"].known_tools
 
 
 def test_transport_configs_preserve_arguments_environment_headers_and_cwd(tmp_path):
@@ -202,8 +202,8 @@ async def test_model_receives_only_current_session_mcp_tools(
 
     seen = []
 
-    def run(self, *, tools, **kwargs):
-        seen.append({tool.name for tool in tools})
+    def run(self, *, tools, system_prompt, **kwargs):
+        seen.append(({tool.name for tool in tools}, system_prompt))
 
         async def events():
             yield StreamEvent("text", {"delta": "ok"})
@@ -228,8 +228,14 @@ async def events():
                 session_id=session.session_id, prompt=[TextContentBlock(text="hello")]
             )
         # Bub aliases dots for the model, but execution still uses each session's tool object.
-        assert "mcp_first_echo" in seen[0] and "mcp_second_echo" not in seen[0]
-        assert "mcp_second_echo" in seen[1] and "mcp_first_echo" not in seen[1]
+        for (names, prompt), own, other in zip(
+            seen,
+            ["mcp_first_echo", "mcp_second_echo"],
+            ["mcp_second_echo", "mcp_first_echo"],
+            strict=True,
+        ):
+            assert "mcp_describe" in names and not {own, other} & names
+            assert own in prompt and other not in prompt
     finally:
         await agent.shutdown()
 
@@ -474,7 +480,7 @@ async def test_session_tools_override_configured_mcp_and_restore_on_close(
         assert len(clients[1].calls) == 2
         runtime_agent = agent._runtime_agents[session.session_id]
         await agent.close_session(session.session_id)
-        assert await runtime_agent.tools["mcp.shared_echo"].run() == "configured"
+        assert await runtime_agent.known_tools["mcp.shared_echo"].run() == "configured"
         await configured._manager.stop()
         assert "mcp.shared_echo" not in runtime_agent.tools
     finally:

packages/bub-acp-server/tests/test_subagents.py

@@ -73,6 +73,7 @@ def __init__(self, events: list[StreamEvent], *, error: Exception | None = None)
         self.error = error
         self.calls: list[dict[str, Any]] = []
         self.tools = {"subagent": REGISTRY["subagent"], "bash": REGISTRY["bash"]}
+        self.known_tools = self.tools
 
     async def run_stream(self, **kwargs: Any) -> AsyncStreamEvents:
         self.calls.append(kwargs)

packages/bub-mcp/README.md

@@ -32,6 +32,22 @@ The file must contain a top-level `mcpServers` mapping:
 }
 ```
 
+## Tool definitions and selection
+
+In ordinary mode, MCP tools start as names and short summaries. `mcp_describe`
+exposes selected native definitions on the next model call. Load missing tools
+from any source together in one call; their original handlers remain unchanged.
+Definitions follow first-discovery order in each tape context and survive restart.
+New sessions, reset and handoff start fresh. Summaries stay stable as definitions
+load. Builtins remain directly available; code mode uses its complete allowed stub.
+
+Configure `mcp.allowed_tools` and `mcp.excluded_tools` in Bub's `config.yml`, or set
+`BUB_MCP_ALLOWED_TOOLS` and `BUB_MCP_EXCLUDED_TOOLS` to JSON arrays. Patterns accept
+runtime names (`mcp.notes_lookup`), model aliases (`mcp_notes_lookup`) and shell
+wildcards. Exclusions win; an omitted allowlist allows all, `[]` allows none.
+Restart Bub after changing configuration. Filtering limits discovery, native calls
+and code mode; server discovery and listing remain unchanged.
+
 ## CLI Usage
 
 Use the CLI to inspect and manage `mcp.json`:
@@ -97,12 +113,14 @@ finally:
 Discovered tools belong to the channel and are available through `channel.tools`; discovery does
 not modify `bub.tools.REGISTRY`. Bub snapshots that registry when an `Agent` is created, so changing
 it afterward would not update existing agents. The plugin's `load_state` hook waits for startup
-discovery and binds tools to the turn's Agent (including an explicit `_runtime_agent`). Stopping
-the channel removes its bindings and restores any tools it replaced.
+discovery and binds the inventory and provider to the turn's Agent (including an explicit `_runtime_agent`). Remote
+tools are selected per request without changing `Agent.tools`; comma commands resolve them
+through the discovery inventory. Code mode loads the allowed catalog into its complete stub.
+Stopping the channel removes its inventory, revealing any tools shadowed by that source.
 
 An embedding application can call `channel.bind_agent(agent)` after startup discovery completes,
 or `await channel.bind_runtime_tools(framework, message)` from its own `load_state` hook. Both paths
-use instance tools. This requires Bub's instance-tool API introduced in upstream PR #311.
+use instance tools. This requires Bub's tool-discovery API introduced in upstream PR #339.
 
 The default `MCPChannel()` behavior remains backed by Bub's `mcp.json`. Read-only channels reject
 `add()` and `remove()` so an embedding plugin remains the owner of its source configuration. The

packages/bub-mcp/src/bub_mcp/config.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+from fnmatch import fnmatchcase
 from pathlib import Path
 from typing import Any
 
@@ -28,6 +29,28 @@ class MCPSettings(bub.Settings):
 
     config_path: Path = Field(default_factory=default_config_path)
     init_timeout_seconds: float | None = 20.0
+    allowed_tools: list[str] | None = Field(
+        None,
+        description="Tool name patterns to expose; None allows all, an empty list allows none.",
+    )
+    excluded_tools: list[str] = Field(
+        default_factory=list,
+        description="Tool name patterns to exclude, even when allowed.",
+    )
+
+    def allows_tool(self, name: str) -> bool:
+        """Match runtime names or model aliases without changing remote schemas."""
+        alias = name.replace(".", "_")
+
+        def matches(patterns: list[str]) -> bool:
+            return any(
+                fnmatchcase(name, pattern) or fnmatchcase(alias, pattern)
+                for pattern in patterns
+            )
+
+        return (
+            self.allowed_tools is None or matches(self.allowed_tools)
+        ) and not matches(self.excluded_tools)
 
     def read_mcp_servers(self) -> dict[str, Any]:
         return read_config(self.config_path)

packages/bub-mcp/src/bub_mcp/plugin.py

@@ -7,7 +7,7 @@
 from copy import deepcopy
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Literal, Self, TypedDict, final
-from weakref import WeakKeyDictionary
+from weakref import WeakSet
 
 import fastmcp
 import mcp.types
@@ -18,6 +18,7 @@
 from bub.tools import Tool, ToolContext
 from bub.channels.contracts import MessageHandler
 from bub.envelope import Envelope, field_of
+from bub.tape import Tape
 from bub.turn import TurnState
 from loguru import logger
 
@@ -28,6 +29,8 @@
     from bub.framework import BubFramework
 
 TOOL_PREFIX = "mcp."
+MCP_TOOLS_STATE_KEY = "_mcp_tools"
+DEFINITIONS_LOADED_EVENT = "mcp.definitions.loaded"
 TEXT_OUTPUT_SCHEMA: dict[str, Any] = {"type": "string"}
 
 type ResultMode = Literal["text", "structured", "wrapped"]
@@ -159,9 +162,7 @@ def __init__(self) -> None:
         self._lock = asyncio.Lock()
         self._bootstrap_task: asyncio.Task[None] | None = None
         self._servers: dict[str, MCPServerState] = {}
-        self._bindings: WeakKeyDictionary[
-            Agent, dict[str, tuple[Tool, Tool | None]]
-        ] = WeakKeyDictionary()
+        self._agents: WeakSet[Agent] = WeakSet()
         self._stop_event: asyncio.Event | None = None
 
     @classmethod
@@ -210,9 +211,13 @@ async def stop(self) -> None:
                 server.client = None
                 server.connected = False
 
-        for agent, bindings in list(self._bindings.items()):
-            self._restore_tools(agent, bindings)
-        self._bindings.clear()
+        for agent in self._agents:
+            agent.tool_sources.pop(self, None)
+            if not any(isinstance(source, MCPChannel) for source in agent.tool_sources):
+                agent.tool_providers.remove(prepare_mcp_tools)
+                if agent.tools.get(mcp_describe.name) is mcp_describe:
+                    agent.tools.pop(mcp_describe.name)
+        self._agents.clear()
 
         for client in clients:
             await self._close_client(client)
@@ -225,35 +230,17 @@ def tools(self) -> dict[str, Tool]:
             for server in self._servers.values()
             if server.connected
             for tool in server.tools
+            if self.settings.allows_tool(tool.name)
         }
 
     def bind_agent(self, agent: Agent) -> None:
-        """Refresh this channel's tools on an Agent, preserving name collisions."""
-        previous = self._bindings.pop(agent, {})
-        tools = self.tools
-        self._restore_tools(
-            agent,
-            {name: binding for name, binding in previous.items() if name not in tools},
-        )
-        bindings = {}
-        for name, remote_tool in tools.items():
-            original = previous[name][1] if name in previous else agent.tools.get(name)
-            bindings[name] = (remote_tool, original)
-            agent.tools[name] = remote_tool
-        if bindings:
-            self._bindings[agent] = bindings
-
-    @staticmethod
-    def _restore_tools(
-        agent: Agent, bindings: dict[str, tuple[Tool, Tool | None]]
-    ) -> None:
-        for name, (installed, original) in bindings.items():
-            if agent.tools.get(name) is not installed:
-                continue
-            if original is None:
-                agent.tools.pop(name, None)
-            else:
-                agent.tools[name] = original
+        """Register discovery definitions separately from request preparation."""
+        self._agents.add(agent)
+        # Explicit tool sets may omit the globally registered discovery helper.
+        agent.tools.setdefault(mcp_describe.name, mcp_describe)
+        agent.tool_sources[self] = self.tools
+        if prepare_mcp_tools not in agent.tool_providers:
+            agent.tool_providers.append(prepare_mcp_tools)
 
     async def bind_runtime_tools(
         self, framework: BubFramework, message: Envelope
@@ -542,3 +529,65 @@ def mcp_list(*, context: ToolContext) -> MCPServerList:
             for name, server in manager.list().items()
         }
     }
+
+
+@tool(name="mcp.describe", context=True, preserve=True)
+async def mcp_describe(names: list[str], *, context: ToolContext) -> str:
+    """Expose complete native definitions for selected MCP tools on the next model call.
+
+    Use exact names from the mcp_tools catalog. Call already available tools directly.
+    Load the missing definitions needed for the current task together in one call.
+    """
+    from bub.builtin.tools import resolve_tool_names
+
+    available: dict[str, Tool] = context.state.get(MCP_TOOLS_STATE_KEY, {})
+    resolved = resolve_tool_names(names, all_names=available)
+    if not resolved:
+        raise ValueError("provide at least one available MCP tool name")
+    await context.tape.append_event(
+        DEFINITIONS_LOADED_EVENT, {"names": sorted(resolved)}, context=False
+    )
+    aliases = ", ".join(name.replace(".", "_") for name in sorted(resolved))
+    return f"Complete native definitions are now available for: {aliases}. Call these tools directly."
+
+
+async def prepare_mcp_tools(tools: list[Tool], tape: Tape) -> tuple[list[Tool], str]:
+    """Prepare scoped MCP tools across sources in the tape's discovery order."""
+    agent = tape.context.state["_runtime_agent"]
+    known = agent.known_tools
+    owned = {
+        name: item
+        for source, definitions in agent.tool_sources.items()
+        if isinstance(source, MCPChannel)
+        for name, item in definitions.items()
+        if known.get(name) is item
+    }
+    available = {
+        item.name: item for item in tools if item.name in owned and item.agent_use
+    }
+    native = [
+        item
+        for item in tools
+        if item.name not in owned and item.name != mcp_describe.name
+    ]
+    tape.context.state[MCP_TOOLS_STATE_KEY] = available
+    code_mode = tape.context.state.get("code_mode") and any(
+        item.name == "run_code" for item in native
+    )
+    if code_mode:
+        return [*native, *available.values()], ""
+    if available:
+        native.append(mcp_describe)
+    loaded: dict[str, None] = {}
+    query = tape.context.build_query(tape.query()).kinds("event")
+    for entry in await tape.store.fetch_all(query):
+        if entry.payload.get("name") == DEFINITIONS_LOADED_EVENT:
+            loaded.update(dict.fromkeys(entry.payload["data"]["names"]))
+    native.extend(available[name] for name in loaded if name in available)
+    lines = []
+    for name, item in available.items():
+        summary = item.description.strip().split("\n", 1)[0].split(". ", 1)[0][:180]
+        lines.append(
+            f"- {name.replace('.', '_')}" + (f": {summary}" if summary else "")
+        )
+    return native, f"<mcp_tools>\n{'\n'.join(lines)}\n</mcp_tools>" if lines else ""