McpExtension#

Build extensions, choose hooks, and manage request-scoped state. See the guide.

class zett_agent.extensions.mcp.McpExtension(servers=None, *, config_path=None, server_keys=('servers', 'mcpServers'), namespace_tools=True, client_factory=None, strict_config=True, connect_timeout=30.0, call_timeout=300.0, tools_cache_ttl=300.0)[源代码]#

Discover and register namespaced tools from configured MCP servers.

Connections are request-scoped: they open during on_tool and remain alive while the model/tool loop runs. Success, failure, and cancellation all close every transport. Remote tool search from server docs is exposed as docs__search by default, preventing ambiguous registrations when several servers provide the same tool name.

参数:

  • servers (Sequence[McpServer] | None) -- Explicit server definitions, merged after the file's servers.

  • config_path (Path | None) -- JSON file to load; re-read whenever it changes.

  • server_keys (Sequence[str]) -- Accepted root keys of the JSON file.

  • namespace_tools (bool) -- Expose tools as <server>__<tool>.

  • client_factory (McpClientFactory | None) -- Replaces the SDK client, mainly for tests.

  • strict_config (bool) -- When true (default) one invalid server entry makes the whole configuration invalid; when false it is skipped and logged so the remaining servers still work.

  • connect_timeout (float | None) -- Seconds allowed to connect and list one server.

  • call_timeout (float | None) -- Seconds allowed for one tool call.

  • tools_cache_ttl (float) -- Seconds a server's tool list is reused; 0 disables.

示例

Usage:

# With no arguments, load Streamable HTTP servers from
# ~/.zett/mcp.json when that file exists. A configured path that
# does not exist yet is treated as no servers and is still named in
# the request's system message.
extension = McpExtension()

# A custom path and explicit server definitions may be used
# independently or merged.
extension = McpExtension(
    servers=[
        McpHttpServer(name="docs", url="http://127.0.0.1:8000/mcp"),
        McpStdioServer(name="local", command="python", args=("server.py",)),
    ],
    config_path="./mcp.json",
    server_keys=("servers", "mcpServers"),
)

mcp.json accepts either servers or the ecosystem-compatible mcpServers key. Streamable HTTP and local stdio transports may be declared together:

{
  "servers": {
    "docs": {
      "type": "streamable-http",
      "url": "https://docs.example.com/mcp",
      "headers": {
        "Authorization": "Bearer token",
        "X-Tenant-ID": "tenant-1"
      }
    },
    "local": {
      "command": "uvx",
      "args": ["some-mcp-server"],
      "env": {"TOKEN": "secret"},
      "cwd": "/path/to/project"
    }
  }
}
property servers: tuple[McpServer, ...]#

Servers declared in the configuration file followed by explicit ones.

async on_tool(context)[源代码]#

Discover every server's tools concurrently; connections open lazily.

Tool lists are cached per server configuration, so a warm cache starts no process and opens no socket until the model actually calls a tool. A server that fails or times out is reported to the model and skipped instead of failing the request.

参数:

context (AgentRunContext)

返回类型:

None

async on_state(context)[源代码]#

Describe where MCP servers are configured and how their tools are named.

参数:

context (AgentRunContext)

返回类型:

None

async on_success(context, result)[源代码]#

Close request transports after a successful final answer.

参数:
返回类型:

None

async on_error(context, error)[源代码]#

Close request transports after a failed request.

参数:
返回类型:

None

async on_compact(context)[源代码]#

Close transports of a maintenance pass, which never reaches on_success.

参数:

context (AgentRunContext)

返回类型:

None

async on_event(context, event)[源代码]#

Close request transports when cancellation bypasses on_error.

参数:
返回类型:

None

accept(config, event)#

Handle one external event and report whether it was accepted.

The Agent broadcasts each event to every registered extension. Override this synchronous hook when an extension waits for input from another thread, HTTP request, TUI, or Web UI. The default ignores the event. Config is passed separately and can be None before initialization. Each extension decides whether the supplied identity and payload match its protocol. Return True to include this extension's name in the broadcast result; returning False does not stop later receivers.

参数:
返回类型:

bool

async after_model(context, response)#

Run after the complete assistant message is appended.

参数:
返回类型:

None

async after_run(context, result)#

Run after a successful final answer is appended.

参数:
返回类型:

None

async after_tool(context, call, result, error)#

Inspect or modify a tool result before it is appended.

error is the original execution exception for a failed result and None after successful execution. Changes made to result are included in context, persistence, and the next model request. Raising prevents the result from being appended.

参数:
返回类型:

None

async after_turn(context, result)#

Run once the turn's assistant message and tool results are settled.

result is the assistant message that opened the turn: either the final answer, or the tool-calling message whose ToolMessages were appended. Every requested call is settled first, including calls that failed, and calls that steering skipped, which arrive as unsuccessful or skipped ToolMessages.

Steering that interrupts a turn is accepted while the runtime settles that batch, so its UserMessage is already appended when this hook runs and the steering turn follows as the next turn.

A turn that never completed, because the request failed or was cancelled, has no after_turn call; use on_error and RunCancelledEvent for those paths.

参数:
返回类型:

None

async before_model(context, request)#

Inspect the current request before preprocessing a primary model call.

Request fields are frozen. Change messages through AgentRunContext.add_message() or AgentRunContext.replace_messages(), local tools through context.tools, and provider tools through context.server_tools; the runtime rebuilds these fields before the next hook and before calling the provider. This is a shallow request view, not an immutable copy of each message or tool definition.

示例

Inspect the tools offered in this iteration:

async def before_model(self, context, request):
    offered = {definition.name for definition in request.tools}
    if "write_file" in offered:
        context.add_message(
            SystemMessage(content="Read existing files before editing."),
        )
参数:
返回类型:

None

async before_run(context)#

Run after the transformed user input is appended and published.

参数:

context (AgentRunContext)

返回类型:

None

async before_tool(context, call)#

Run before one requested tool is invoked.

参数:
返回类型:

None

async before_turn(context)#

Run before this turn's model call.

Runs after before_run() on the first turn and after the previous turn's after_turn() on later ones. Model request assembly, compaction, and before_model() follow this hook, so context edited here reaches the provider.

参数:

context (AgentRunContext)

返回类型:

None

property name: str#

Unique registration identity, defaulting to the concrete class name.

Override with a class attribute or assign an instance name to register independently configured instances of the same extension type.

async on_message(context)#

Transform context.input_message after state restoration, before append.

Replace the pending UserMessage to implement externally selected modes. Do not append or persist it here: the runtime appends the final message exactly once after all input hooks finish. before_run sees that message through AgentRunContext.add_message() or AgentRunContext.replace_messages(), and persistence subscribers have already received it.

参数:

context (AgentRunContext)

返回类型:

None

async on_model_request(context, request, call_next)#

Wrap one provider request and its complete streamed response.

Pass request (or a dataclass replacement) to call_next. The default closes the inner stream so cancellation reaches the provider.

参数:
返回类型:

AsyncIterator[ModelEvent]

async on_tool_call(context, call, call_next)#

Wrap one registered local tool handler and return its ToolResult.

An extension may adjust call.arguments, return without calling the next layer, transform its result, or raise into normal tool-failure handling.

参数:
返回类型:

ToolResult

property pending: bool#

Return whether this extension is participating in an active request.

The Agent updates this state for every registered extension, including extensions whose hooks are currently idle. One extension instance may be shared by multiple Agents and sessions; pending remains true until all requests using that instance have finished, failed, or been cancelled. Persistent data retained between requests does not count.