A guarded tool gateway seam so channels can run the agent's tools

[ekarad1um]

Problem / Motivation

A channel is constructed with only (config, bus), and the bus carries text. So a channel has no path to run the agent's tools, even when it legitimately needs to. One of the motivating case is an end-to-end / real-time voice channel, e.g. an external model emits a function_call and the channel must execute it against nanobot's ToolRegistry so allow-lists / SSRF / workspace / file-state guards match every other turn. Not proposing voice in core, that stays an external channel plugin, but there is currently no supported way to do that.

Proposed Solution

Expose a narrow ToolGateway protocol that opt-in channels receive:

# nanobot/channels/base.py  (or a shared module)
class ToolGateway(Protocol):
    def get_tool_definitions(self) -> list[dict]: ...
    async def execute_tool(self, session_key: str, name: str, args: Any, *,
                           channel: str, chat_id: str | None = None) -> Any: ...

class ChannelBase:
    wants_tool_gateway: bool = False   # opt in to receive a tool_gateway= kwarg

AgentLoop satisfies it structurally with two additive methods:

# nanobot/agent/loop.py
def get_tool_definitions(self) -> list[dict]:
    """OpenAI-style schemas for the live tool registry (== the runner's)."""
    return self.tools.get_definitions()

async def execute_tool(self, session_key: str, name: str, args: Any, *,
                       channel: str, chat_id: str | None = None) -> Any:
    """Execute one tool with the SAME guards as a normal turn.

    Binds the identical workspace / request / file-state contextvars as
    _run_agent_loop and resets them in finally, so there is no second,
    unguarded execution path. args is a dict or JSON string.
    """
    session = self.sessions.get_or_create(session_key)
    scope = self.workspace_scopes.for_turn(
        channel=channel, message_metadata=None,
        session_metadata=session.metadata if session is not None else None,
    )
    ctx = RequestContext(channel=channel, chat_id=chat_id or session_key,
                         message_id=None, session_key=session_key, metadata={})
    fs = bind_file_states(self._file_state_store.for_session(session_key))
    rq = bind_request_context(ctx)
    ws = bind_workspace_scope(scope)
    try:
        return await self.tools.execute(name, args)
    finally:
        reset_workspace_scope(ws); reset_request_context(rq); reset_file_states(fs)

Wiring is opt-in and mirrors the existing session_manager / cron_service kwargs:

# channels/manager.py
if getattr(cls, "wants_tool_gateway", False) and self._agent_loop is not None:
    kwargs["tool_gateway"] = self._agent_loop
# cli/commands.py — pass agent_loop=agent into ChannelManager(...)

Alternatives Considered

Safety:

• No new execution path: the only path stays ToolRegistry.execute. execute_tool binds the same three contextvars as _run_agent_loop and resets them in finally, so workspace / SSRF / file-state enforcement is byte-for-byte a normal turn.
• No tool gains any capability: SSRF loopback stays gated on source_channel=="websocket", so a voice channel gets less access.
• Zero blast radius: existing channels don't set wants_tool_gateway, so nothing changes for them. The kwarg is only handed to a channel that opts in and only when the app wires agent_loop=.
• Narrow surface: channels receive a 2-method facade, not the whole loop.

MCP also needs the same guard binding, adds a transport hop, but doesn't fit the realtime function_call -> function_call_output exchange. The seam is the smaller primitive.

Related: #4010

Related Component

Channel (WeChat, Feishu, Telegram, etc.)

Additional Context

No response