Agent Harness 工程：实现 MCP Server 与 Client

上一篇为每个 Task 创建了独立 Git worktree。文件工具、Shell、依赖和测试产物终于拥有了明确的任务工作区，但这些工具仍然直接注册在 Harness 进程里：

Agent Loop -> Tool Registry -> Python function

这种结构适合从零验证 Agent Loop。随着能力增多，问题也会逐渐显现：

工具实现与 Harness 使用同一进程、同一依赖环境；
每接入一个外部服务，都要编写一套专用发现与调用逻辑；
独立服务和其它 Agent 很难复用已有工具；
工具升级、退出或崩溃会直接影响主进程；
Tool Registry 需要同时理解本地函数、HTTP API 和各种第三方协议。

MCP（Model Context Protocol）为能力提供者和 Agent Harness 之间增加了一层协议边界：

Agent Loop -> Tool Registry -> MCP Client <=> MCP Server -> real capability

本章按 MCP 2025-11-25 规范实现一条完整但克制的工具链：

用官方 Python SDK 编写一个最小 MCP Server；
由 Harness 启动本地 Server，并完成初始化握手；
通过 tools/list 发现工具并转换成模型可见的 schema；
将模型的 tool_call 映射成 MCP tools/call；
校验输入和结构化输出，区分工具错误与连接故障；
继续复用前面实现的 Hook、权限、Task 与 Worktree。

重点不是手写 JSON-RPC，而是把 MCP 放到 Harness 中正确的位置。

先分清四个角色

MCP 的名字里虽然有 Model，但模型通常不会直接连接 MCP Server。

在当前架构中，四个角色分别是：

角色	职责
Model	根据提示词和工具 schema 生成 provider 格式的 `tool_call`
Agent Harness / Host	维护 Agent Loop、消息、权限、Task、Worktree，并决定暴露哪些能力
MCP Client	与一个 MCP Server 建立连接，完成协议协商、能力发现和请求转发
MCP Server	通过 MCP 暴露 Tools、Resources 或 Prompts，并执行背后的真实能力

一个 Host 可以同时持有多个 Client，每个 Client 对应一个 Server。Client 是 Host 内部的协议组件，不是另一个 Agent。

模型仍然只生成普通工具调用；Harness 负责权限、名称映射和上下文注入，MCP Client 只负责协议通信。

MCP Server 可以暴露三类常见能力：

Tools：可执行动作，由模型或 Host 发起调用；
Resources：可读取的上下文数据，通过 URI 标识；
Prompts：Server 提供的提示词模板。

本章只接入 Tools。Resources 和 Prompts 不是“另一种工具名”，它们有各自的发现与读取协议，不应该为了省代码全部伪装成 tools/call。

还有三个容易混淆的边界：

MCP 的 Tasks 能力不等于第 13 章实现的持久化 Task Graph；
MCP 的 Roots 用于向 Server 表达可用文件根，不是文件系统沙箱；
MCP 统一了能力交换协议，但不会自动提供审批、权限或副作用幂等。

这些工程责任仍然属于 Harness。

固定规范与 SDK 范围

MCP 规范版本和 SDK 包版本是两套编号。本文固定协议版本 2025-11-25，并使用官方 Python SDK 的稳定发布范围：

pip install "mcp>=1.27,<2" "jsonschema>=4.20,<5"

给依赖设置上界很重要。协议可能保持兼容，SDK 的 Python API 却可能在下一个主版本调整；没有上界的自动安装会让同一份示例在未来得到不同接口。

实际项目还应该把解析后的精确版本写进 lockfile。本文不自己实现 JSON-RPC 帧、请求 ID、初始化通知和进程退出，因为这些都由官方 SDK 负责。

为什么先使用 stdio

当前规范定义了两种标准传输：

传输	适用场景
`stdio`	Harness 在本机启动受控子进程，工具跟随任务
Streamable HTTP	连接远程或共享服务，需要认证与网络治理

本章先使用 stdio，因为它与上一章的 Worktree 很自然地配合：

Harness 为 Task 创建 worktree；
Harness 以该 worktree 作为 cwd 启动 MCP Server；
Server 生命周期跟随 Task Session；
连接不需要额外监听端口。

在 stdio 传输中：

Client 启动 Server 子进程；
Client 向子进程的 stdin 写入 MCP 消息；
Server 只在 stdout 写入 MCP 消息；
日志必须写到 stderr，否则会破坏协议流。

这也是为什么不要在 Server 里随手 print("server started")。对人类友好的那行日志，对 MCP Client 来说是一条无法解析的 JSON-RPC 消息。

实现一个 Worktree 文件 Server

先创建 workspace_server.py。它只暴露一个 read_file，但已经包含真实工具需要的几个关键约束：

import os
from pathlib import Path
from typing import Annotated

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field


MAX_FILE_BYTES = 2_000_000
WORKSPACE_ROOT = Path(
    os.environ["AGENT_WORKSPACE_ROOT"]
).resolve(strict=True)

mcp = FastMCP(
    "workspace-tools",
    instructions="Tools operate inside the assigned task workspace.",
)


class FilePreview(BaseModel):
    path: str
    content: str
    truncated: bool


def resolve_workspace_path(relative_path: str) -> Path:
    candidate = Path(relative_path)
    if candidate.is_absolute():
        raise ValueError("absolute paths are not allowed")

    resolved = (WORKSPACE_ROOT / candidate).resolve(strict=True)
    if (
        resolved != WORKSPACE_ROOT
        and WORKSPACE_ROOT not in resolved.parents
    ):
        raise ValueError("path escapes task workspace")
    return resolved


@mcp.tool()
def read_file(
    path: str,
    line_limit: Annotated[int, Field(ge=1, le=1000)] = 200,
) -> FilePreview:
    """Read a UTF-8 text file in the assigned task workspace."""
    target = resolve_workspace_path(path)
    if not target.is_file():
        raise ValueError("path is not a regular file")
    if target.stat().st_size > MAX_FILE_BYTES:
        raise ValueError("file is too large")

    lines: list[str] = []
    truncated = False
    with target.open("r", encoding="utf-8") as handle:
        for line_number, line in enumerate(handle):
            if line_number >= line_limit:
                truncated = True
                break
            lines.append(line.rstrip("\r\n"))

    return FilePreview(
        path=str(target.relative_to(WORKSPACE_ROOT)),
        content="\n".join(lines),
        truncated=truncated,
    )


if __name__ == "__main__":
    mcp.run(transport="stdio")

FastMCP 会根据函数签名生成 inputSchema。FilePreview 则让 SDK 同时生成 outputSchema，并把返回值放进 structuredContent。

如果工具抛出 ValueError，Server 不会让整个连接退出，而是返回 isError=true 的 CallToolResult。这类错误可以写回模型，让它修正路径后继续调用。

这里的路径校验用于阻止明显的绝对路径、.. 和已有 symlink 越界，但它仍然不是完整沙箱。Server 进程必须运行在上一章的任务沙箱里，只把当前 worktree 挂载为可写；否则它仍可能直接读取用户目录或调用其它系统 API。

Server 配置只能来自 Harness

模型不能决定启动哪个可执行文件、使用哪个工作目录，或者向子进程注入哪些环境变量。先定义一份由 Harness Policy 管理的配置：

from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class McpServerConfig:
    id: str
    command: str
    args: tuple[str, ...]
    cwd: Path
    env: dict[str, str]
    allowed_tools: frozenset[str]
    supported_protocol_versions: frozenset[str]
    request_timeout_seconds: float = 30.0

本地 workspace Server 的配置由当前 Task Workspace 派生：

import sys
from pathlib import Path


def build_workspace_server_config(
    workspace: TaskWorkspace,
    server_script: Path,
) -> McpServerConfig:
    root = workspace.root.resolve(strict=True)
    temp_dir = workspace.temp_dir.resolve(strict=True)
    runtime_home = temp_dir / "home"
    runtime_home.mkdir(parents=True, exist_ok=True)

    return McpServerConfig(
        id=f"workspace-{workspace.task_id}",
        command=sys.executable,
        args=(str(server_script.resolve(strict=True)),),
        cwd=root,
        env={
            "AGENT_WORKSPACE_ROOT": str(root),
            "HOME": str(runtime_home),
            "PYTHONUNBUFFERED": "1",
        },
        allowed_tools=frozenset({"read_file"}),
        supported_protocol_versions=frozenset({"2025-11-25"}),
        request_timeout_seconds=30,
    )

这里有四层限制：

command 和脚本路径来自部署配置，不来自模型参数；
cwd 与 AGENT_WORKSPACE_ROOT 都绑定当前 Task；
allowed_tools 是拒绝优先的显式 allowlist；
环境变量只添加必要项，并把 HOME 指向任务 Runtime Root。

官方 stdio Client 会继承一小组基础环境变量，再叠加这里的 env，不会直接复制全部进程环境。但环境变量过滤仍不能替代进程沙箱：凭据文件、网络和系统调用还需要操作系统边界控制。

正确完成初始化生命周期

MCP 连接不能一启动就调用 tools/list。根据生命周期规范，第一条交互必须是初始化：

Client -> initialize
Server -> protocolVersion + capabilities + serverInfo
Client -> notifications/initialized

官方 ClientSession.initialize() 会完成这三步，并拒绝 SDK 不支持的协议版本。Harness 还可以增加自己的部署版本 allowlist。

下面用 AsyncExitStack 管理子进程、stdio 流和 Client Session：

import asyncio
from contextlib import AsyncExitStack
from datetime import timedelta

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import (
    Implementation,
    ServerNotification,
    ToolListChangedNotification,
)


class McpContractError(RuntimeError):
    pass


class McpConnection:
    def __init__(self, config: McpServerConfig) -> None:
        self.config = config
        self.catalog_dirty = asyncio.Event()
        self._stack: AsyncExitStack | None = None
        self._session: ClientSession | None = None
        self.server_info: Implementation | None = None

    @property
    def session(self) -> ClientSession:
        if self._session is None:
            raise RuntimeError("MCP connection is not open")
        return self._session

    async def _handle_message(self, message: object) -> None:
        if (
            isinstance(message, ServerNotification)
            and isinstance(
                message.root,
                ToolListChangedNotification,
            )
        ):
            self.catalog_dirty.set()

    async def __aenter__(self) -> "McpConnection":
        if self._stack is not None:
            raise RuntimeError("MCP connection is already open")
        if not self.config.cwd.resolve(strict=True).is_dir():
            raise ValueError("MCP server cwd is not a directory")

        stack = AsyncExitStack()
        self._stack = stack
        try:
            parameters = StdioServerParameters(
                command=self.config.command,
                args=list(self.config.args),
                cwd=self.config.cwd,
                env=dict(self.config.env),
            )
            read_stream, write_stream = (
                await stack.enter_async_context(
                    stdio_client(parameters)
                )
            )
            session = await stack.enter_async_context(
                ClientSession(
                    read_stream,
                    write_stream,
                    read_timeout_seconds=timedelta(
                        seconds=self.config.request_timeout_seconds
                    ),
                    message_handler=self._handle_message,
                    client_info=Implementation(
                        name="agent-harness",
                        version="0.1.0",
                    ),
                )
            )

            initialization = await session.initialize()
            protocol_version = str(
                initialization.protocolVersion
            )
            if (
                protocol_version
                not in self.config.supported_protocol_versions
            ):
                raise McpContractError(
                    "negotiated protocol version is not allowed"
                )
            if initialization.capabilities.tools is None:
                raise McpContractError(
                    "server does not advertise tools"
                )

            self._session = session
            self.server_info = initialization.serverInfo
            return self
        except BaseException:
            self._session = None
            self.server_info = None
            self._stack = None
            await stack.aclose()
            raise

    async def __aexit__(self, *exc_info: object) -> None:
        stack = self._stack
        self._session = None
        self.server_info = None
        self._stack = None
        if stack is not None:
            await stack.aclose()

stdio_client() 退出时会先关闭子进程的 stdin，等待 Server 自行退出；超时后再终止进程。不要再额外维护一套互相竞争的 subprocess 清理逻辑。

连接应该在同一个 Harness 生命周期任务中打开和关闭。不要在一个异步任务中进入 stdio context，又让另一个任务随意执行 __aexit__()；底层 task group 和取消域需要稳定的所有者。

分页发现工具，而不是硬编码名称

初始化只告诉 Client“Server 支持 Tools”，不会直接返回工具清单。Client 还要调用 tools/list。

工具发现至少要处理：

nextCursor 分页；
Server 内重复名称；
Harness allowlist；
JSON Schema 合法性；
多个 Server 之间的名称冲突；
Server 返回的超长描述和异常目录。

先准备 JSON Schema Validator 和稳定的本地工具名：

import hashlib
import re
from typing import Any

from jsonschema import Draft202012Validator
from jsonschema.protocols import Validator
from jsonschema.validators import validator_for


LOCAL_TOOL_NAME_LIMIT = 60


def build_validator(
    schema: dict[str, Any],
) -> Validator:
    validator_class = validator_for(
        schema,
        default=Draft202012Validator,
    )
    validator_class.check_schema(schema)
    return validator_class(schema)


def make_local_tool_name(
    server_id: str,
    remote_name: str,
) -> str:
    source = f"mcp_{server_id}_{remote_name}"
    slug = re.sub(
        r"[^A-Za-z0-9_-]+",
        "_",
        source,
    ).strip("_")
    digest = hashlib.sha256(
        f"{server_id}\0{remote_name}".encode("utf-8")
    ).hexdigest()[:8]
    prefix_length = LOCAL_TOOL_NAME_LIMIT - len(digest) - 1
    return f"{slug[:prefix_length]}_{digest}"

模型最终看到的可能是：

mcp_workspace-task-api_read_file_a134c92e

它不如 read_file 短，但有三个好处：

不同 Server 的同名工具不会互相覆盖；
MCP 名称中的特殊字符不会破坏模型提供方的工具名约束；
截断后仍有稳定摘要，Harness 可以可靠反查远端名称。

不要让模型自己拼接 Server ID 和远端工具名。名称映射必须保存在 Harness 中。

每个发现到的工具保存为一条 Binding：

from copy import deepcopy
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class McpToolBinding:
    local_name: str
    server_id: str
    remote_name: str
    description: str
    input_schema: dict[str, Any]
    output_schema: dict[str, Any] | None
    input_validator: Validator
    output_validator: Validator | None

    def provider_schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.local_name,
                "description": self.description,
                "parameters": deepcopy(self.input_schema),
            },
        }

这里使用的是前几章采用的 OpenAI 风格工具外层结构。MCP inputSchema 与模型提供方的参数 schema 都以 JSON Schema 为基础，但提供方可能只支持其中一个子集。接入任意第三方 Server 时，还要在部署阶段执行 Provider Schema 兼容性检查；遇到不支持的关键字应该拒绝该工具，而不是静默删除约束，造成“模型看到的 schema”和“Harness 实际校验的 schema”不一致。

然后完整遍历 tools/list：

from copy import deepcopy

from mcp.types import PaginatedRequestParams


MAX_TOOL_PAGES = 100
MAX_DISCOVERED_TOOLS = 1000


async def discover_tools(
    connection: McpConnection,
) -> list[McpToolBinding]:
    bindings: list[McpToolBinding] = []
    remote_names: set[str] = set()
    seen_cursors: set[str] = set()
    cursor: str | None = None

    for _ in range(MAX_TOOL_PAGES):
        params = (
            PaginatedRequestParams(cursor=cursor)
            if cursor is not None
            else None
        )
        page = await connection.session.list_tools(
            params=params
        )

        for tool in page.tools:
            if tool.name in remote_names:
                raise McpContractError(
                    f"duplicate MCP tool name: {tool.name}"
                )
            remote_names.add(tool.name)
            if len(remote_names) > MAX_DISCOVERED_TOOLS:
                raise McpContractError(
                    "MCP tool catalog is too large"
                )

            if (
                tool.name
                not in connection.config.allowed_tools
            ):
                continue

            input_schema = deepcopy(tool.inputSchema)
            if input_schema.get("type") != "object":
                raise McpContractError(
                    f"{tool.name} input schema "
                    "must describe an object"
                )

            output_schema = (
                deepcopy(tool.outputSchema)
                if tool.outputSchema is not None
                else None
            )
            bindings.append(
                McpToolBinding(
                    local_name=make_local_tool_name(
                        connection.config.id,
                        tool.name,
                    ),
                    server_id=connection.config.id,
                    remote_name=tool.name,
                    description=(
                        tool.description
                        or f"MCP tool {tool.name}"
                    )[:2000],
                    input_schema=input_schema,
                    output_schema=output_schema,
                    input_validator=build_validator(
                        input_schema
                    ),
                    output_validator=(
                        build_validator(output_schema)
                        if output_schema is not None
                        else None
                    ),
                )
            )

        next_cursor = page.nextCursor
        if next_cursor is None:
            break
        if next_cursor in seen_cursors:
            raise McpContractError(
                "MCP tools/list cursor repeated"
            )
        seen_cursors.add(next_cursor)
        cursor = next_cursor
    else:
        raise McpContractError(
            "MCP tools/list exceeded page limit"
        )

    missing_tools = (
        connection.config.allowed_tools - remote_names
    )
    if missing_tools:
        missing = ", ".join(sorted(missing_tools))
        raise McpContractError(
            f"configured MCP tools are missing: {missing}"
        )

    connection.catalog_dirty.clear()
    return bindings

allowed_tools 与工具发现是两件事：

发现回答“Server 声称自己有什么”；
allowlist 回答“Harness 管理员允许当前部署使用什么”。

不能把 Server 返回的所有工具自动暴露给模型。一个日历 Server 可能同时提供只读查询和删除全部事件，协议发现成功不等于安全策略批准。

当 Server 数量和工具数量继续增大时，还可以在 allowlist 之上增加 Task capability 或 Skill 驱动的本轮激活集合。发现目录、允许目录、模型当前可见目录应该是三个不同层次。

一次调用跨过两套协议

工具列表准备好以后，模型仍然不会生成 MCP JSON-RPC。一次调用实际经过两层：

Model Provider 返回 tool_call.id + local_name + arguments；
Harness 找到 Binding，MCP SDK 再生成自己的 JSON-RPC request ID。

Provider 的 tool_call_id 用于消息闭环；MCP request ID 只属于 Client 与 Server 的连接，二者不能混为一谈。

调用结果是 CallToolResult，常用字段包括：

字段	含义
`content`	文本、图片、音频、资源链接或嵌入资源等内容块
`structuredContent`	与 `outputSchema` 对应的结构化对象
`isError`	工具执行是否失败；失败内容应反馈给模型
`_meta`	Client 或 Host 使用的附加元数据，不应默认写入模型上下文

先实现只接受文本和结构化对象的适配器：

import json
from datetime import timedelta
from typing import Any

from jsonschema.exceptions import ValidationError
from mcp.types import TextContent


def read_text_blocks(blocks: list[object]) -> str:
    texts: list[str] = []
    for block in blocks:
        if not isinstance(block, TextContent):
            raise McpContractError(
                "non-text MCP content must be stored "
                "as an artifact"
            )
        texts.append(block.text)
    return "\n".join(texts)


class McpToolAdapter:
    def __init__(
        self,
        connection: McpConnection,
        bindings: list[McpToolBinding],
    ) -> None:
        self.connection = connection
        self.bindings = {
            binding.local_name: binding
            for binding in bindings
        }
        if len(self.bindings) != len(bindings):
            raise McpContractError(
                "local MCP tool name collision"
            )

    async def call(
        self,
        local_name: str,
        arguments: dict[str, Any],
    ) -> str:
        binding = self.bindings.get(local_name)
        if binding is None:
            return f"Error: unknown tool {local_name}"

        try:
            binding.input_validator.validate(arguments)
        except ValidationError as exc:
            return (
                "Error: invalid arguments for "
                f"{binding.remote_name} - {exc.message}"
            )

        result = await self.connection.session.call_tool(
            binding.remote_name,
            arguments,
            read_timeout_seconds=timedelta(
                seconds=(
                    self.connection.config
                    .request_timeout_seconds
                )
            ),
        )

        plain_text = read_text_blocks(
            list(result.content)
        )
        if result.isError:
            detail = plain_text
            if (
                not detail
                and result.structuredContent is not None
            ):
                detail = json.dumps(
                    result.structuredContent,
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
            suffix = f"\n{detail}" if detail else ""
            return (
                "Error: MCP tool "
                f"{binding.remote_name} failed{suffix}"
            )

        if binding.output_validator is not None:
            if result.structuredContent is None:
                raise McpContractError(
                    f"{binding.remote_name} omitted "
                    "structuredContent"
                )
            try:
                binding.output_validator.validate(
                    result.structuredContent
                )
            except ValidationError as exc:
                raise McpContractError(
                    f"{binding.remote_name} returned "
                    "invalid structuredContent"
                ) from exc

        if result.structuredContent is not None:
            return json.dumps(
                result.structuredContent,
                ensure_ascii=False,
                separators=(",", ":"),
            )
        return plain_text

这段代码刻意区分了三类结果：

参数不符合 inputSchema：返回普通工具错误，模型可以修正参数；
isError=true：Server 已经正常响应，只是工具执行失败，也写回模型；
连接异常或输出违反协议：抛给 Harness Recovery，不伪装成普通业务结果。

官方 SDK 也会校验带有 outputSchema 的成功结果。这里仍然保存并验证发现时的 schema，是为了让当前 Tool Catalog 快照与当前模型调用保持一致，而不是在调用中途悄悄换成另一版目录。

如果 Server 返回图片、音频或嵌入资源，不能静默丢掉，也不应直接把大段 base64 塞进 messages。更完整的实现应该把二进制内容写入第 14 章的 Artifact Store，再向模型返回 artifact ID、媒体类型、大小和受控读取工具。

接回异步 Tool Registry

MCP 调用是异步 I/O，因此 Tool Registry 的统一执行入口也要变成异步。一个最小注册项可以这样定义：

from collections.abc import Awaitable, Callable
from copy import deepcopy
from dataclasses import dataclass
from functools import partial
from typing import Any


AsyncToolHandler = Callable[
    [dict[str, Any]],
    Awaitable[str],
]


@dataclass
class RegisteredTool:
    schema: dict[str, Any]
    handler: AsyncToolHandler


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, RegisteredTool] = {}

    def register_async(
        self,
        name: str,
        schema: dict[str, Any],
        handler: AsyncToolHandler,
    ) -> None:
        if name in self._tools:
            raise ValueError(f"duplicate tool: {name}")
        self._tools[name] = RegisteredTool(
            schema=deepcopy(schema),
            handler=handler,
        )

    def names(self) -> list[str]:
        return list(self._tools)

    def schemas(self) -> list[dict[str, Any]]:
        return [
            deepcopy(tool.schema)
            for tool in self._tools.values()
        ]

    def fork(self) -> "ToolRegistry":
        child = ToolRegistry()
        child._tools = {
            name: RegisteredTool(
                schema=deepcopy(tool.schema),
                handler=tool.handler,
            )
            for name, tool in self._tools.items()
        }
        return child

    async def call(
        self,
        name: str,
        arguments: dict[str, Any],
    ) -> str:
        tool = self._tools.get(name)
        if tool is None:
            return f"Error: unknown tool {name}"
        return await tool.handler(arguments)


def install_mcp_tools(
    registry: ToolRegistry,
    adapter: McpToolAdapter,
) -> None:
    for binding in adapter.bindings.values():
        registry.register_async(
            name=binding.local_name,
            schema=binding.provider_schema(),
            handler=partial(
                adapter.call,
                binding.local_name,
            ),
        )

前几章的同步本地工具可以先用异步 wrapper 注册；阻塞型文件或子进程操作应放进 asyncio.to_thread()，原生异步工具则直接 await。不要为了兼容旧接口，在事件循环里同步等待 MCP coroutine。

第 5 章的 Hook 流程只需要把真实执行改成 await：

async def run_tool_call_with_hooks(
    state: HarnessState,
    tool_call: Any,
    messages: list[dict[str, Any]],
    step: int,
) -> str:
    tool_name = tool_call.function.name
    arguments, error = parse_tool_arguments(tool_call)
    if error:
        return error

    context = ToolContext(
        step=step,
        tool_call_id=tool_call.id,
        tool_name=tool_name,
        arguments=arguments,
        messages=messages,
    )

    blocked_result = run_before_tool_hooks(context)
    if blocked_result is not None:
        return blocked_result

    context.result = await state.tool_registry.call(
        tool_name,
        arguments,
    )
    run_after_tool_hooks(context)
    return context.result or ""

Agent Loop 写回消息的方式不变：

tool_result = await run_tool_call_with_hooks(
    state,
    tool_call,
    messages,
    step,
)
messages.append(
    {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": tool_result,
    }
)

这正是协议分层的价值：Agent Loop 继续维护 Model Provider 的 tool_call_id，Tool Registry 负责本地名称，MCP SDK 在连接内部维护 JSON-RPC request ID。主循环不需要理解 MCP 消息格式。

权限必须位于 MCP 请求之前

调用顺序应该是：

parse arguments
-> before_tool hooks
-> permission / approval
-> local schema validation
-> MCP tools/call
-> output validation
-> after_tool hooks
-> model tool message

权限检查必须发生在请求发往 Server 之前。否则即使 Harness 最后拒绝把结果写回模型，副作用也可能已经发生。

MCP Tool 可以带有 readOnlyHint、destructiveHint、idempotentHint 等 annotation，但规范明确把它们定位为提示。Harness 不能仅凭远端 Server 自己声明“只读”就免除审批。真正的风险级别应该来自受信任的本地 Policy，annotation 只能作为辅助信息或配置校验信号。

同样，initialize 返回的 Server instructions、工具描述和 _meta 也不应该直接拼进 system prompt。只有管理员批准的 Server 和字段，才能经过长度限制与内容边界后进入 Prompt Assembler。

Tool Catalog 何时刷新

Server 可以声明工具目录支持变化，并发送：

notifications/tools/list_changed

前面的 message_handler 只把 catalog_dirty 设为 true，没有立即改 Tool Registry。这是有意为之。

假设模型刚看到工具 A 的 schema，Server 此时通知目录变化，而 Harness 立刻删除 A、加入 B。模型随后按上一条响应调用 A，就会遇到一个由 Harness 自己制造的竞态。

更稳定的刷新策略是：

一次模型请求开始前固定 Tool Catalog snapshot；
这一轮所有 tool_call 都按同一 snapshot 解析；
当前调用结束后检查 catalog_dirty；
重新执行完整分页发现与 Policy 过滤；
原子替换下一轮使用的 schema、Binding 和名称索引；
将目录版本写入 Trace，便于复现。

如果工具变化会影响正在执行的 Task，还要把 snapshot ID 保存进 checkpoint。目录变化不应该让已恢复任务在没有记录的情况下得到完全不同的能力集合。

与 Task 和 Worktree 对齐

本地 workspace Server 最稳妥的粒度是每个 Task 一个进程和连接：

Task API -> task-api worktree -> workspace MCP Server A
Task UI  -> task-ui worktree  -> workspace MCP Server B

Harness 创建 Task Session 时：

根据 Worktree Record 构造 McpServerConfig；
在任务沙箱中启动 Server；
发现并注册该 Task 允许使用的工具；
Agent Loop 只拿到当前 Session 的 Tool Registry；
Task 结束后先关闭 MCP 连接，再回收 worktree。

模型不需要，也不应该在每次 read_file 时传 workspace_id。工作区属于运行上下文，由 Harness 注入。这样模型即使知道另一个 Task 的 ID，也不能把当前工具切换到对方目录。

与工作区无关的远程 Server，例如只读文档检索，可以在多个 Task 间共享连接。但共享连接要有独立并发上限、租户边界和认证上下文，不能为了节省一个进程而混用不同用户的凭据。

超时之后，副作用可能已经发生

request_timeout_seconds 只限制 Client 等待多久，不证明 Server 没有执行工具。

例如 Harness 调用远端 create_issue：

Server 已经成功创建 Issue；
返回结果前网络中断；
Client 超时；
Harness 看不到 Issue ID。

如果此时自动重试，就可能创建两条 Issue。

因此继续沿用第 12 章的恢复原则：

故障	处理方式
本地参数校验失败	写回模型，让它修正参数
`CallToolResult.isError=true`	写回模型，保留正常连接
JSON-RPC error	按错误类型记录；不要假装成功
超时或连接中断	状态标记为 unknown，先核对副作用
输出缺失或不符合 `outputSchema`	视为 Server contract failure，不把不可信结果交给模型
stdio Server 退出	在安全点重连并重新发现工具，不盲目重放在途写操作

只读且由本地 Policy 确认幂等的工具可以有限重试。写操作仍然需要业务幂等键、Job Record 或查询型补偿工具。远端 annotation 不能单独成为自动重试依据。

MCP SDK 的 request ID 也不是业务幂等键。断线重连后换一个 JSON-RPC ID，不会让 Server 自动识别“这是刚才那次创建操作”。

一条完整的启动路径

把各段代码串起来，单个 Task Session 的结构大致如下：

from dataclasses import replace


async def run_task_agent(
    state: HarnessState,
    workspace: TaskWorkspace,
    server_script: Path,
    user_message: str,
) -> str:
    config = build_workspace_server_config(
        workspace,
        server_script,
    )

    async with McpConnection(config) as connection:
        bindings = await discover_tools(connection)
        mcp_adapter = McpToolAdapter(
            connection,
            bindings,
        )
        task_registry = state.tool_registry.fork()
        install_mcp_tools(
            task_registry,
            mcp_adapter,
        )
        task_state = replace(
            state,
            tool_registry=task_registry,
        )

        return await run_agent_loop(
            task_state,
            user_message,
        )

这里用 fork() 创建 Task Session 自己的 Registry。连接关闭后，task_state 与远端 handler 一起丢弃，基础 state.tool_registry 不会残留指向已关闭 Session 的函数。replace() 来自 dataclasses；如果 HarnessState 不是 dataclass，可以改成显式的 Session State 构造函数。Task 绑定的本地 handler 也应该注册到这份 Session Registry，而不是捕获主工作区状态。

实际 Harness 同时连接多个 Server 时，可以再用一层 AsyncExitStack 统一管理。启动要遵循“全部握手和目录校验成功，再向模型暴露工具”；关闭则反过来：

停止发起新的模型轮次；
等待或按策略取消正在执行的工具调用；
停止后台 Job 对连接的引用；
关闭 MCP Client Session 与 Server 进程；
持久化 Trace、Task checkpoint 和 Artifact；
最后清理 Task worktree。

如果先删除 worktree，再等待 workspace Server 退出，Server 的当前目录和正在读取的文件会突然消失。

常见坑

第一，让模型直接提供 Server command。 这等于把任意进程启动能力交给模型；Server 配置只能来自 Harness allowlist。

第二，自己用 subprocess 和字符串拼 JSON-RPC。 初始化通知、并发请求、取消、超时和退出都容易出现边界错误，应使用官方 SDK。

第三，在 stdio Server 的 stdout 打日志。 stdout 只属于协议，日志写 stderr。

第四，初始化前调用 tools/list。 initialize 必须是第一条协议交互，随后还要发送 notifications/initialized。

第五，只读取第一页工具。 tools/list 支持 cursor 分页；忽略 nextCursor 会得到不完整目录。

第六，把所有发现到的工具直接交给模型。 Discovery 不是授权，至少还需要 Server 和 Tool 两层 allowlist。

第七，多个 Server 的同名工具互相覆盖。 使用稳定本地别名，并保存 local-to-remote Binding。

第八，只相信模型提供方做参数校验。 Harness 在发送请求前仍要按 MCP inputSchema 校验。

第九，忽略 isError。 它表示一次正常完成的协议调用中，工具执行失败；不应因此重启整个连接。

第十，把任何异常都转换成工具文本。 连接断开、输出违反 schema 和 Server contract failure 应进入 Recovery 与审计。

第十一，超时后自动重试写操作。 超时不等于未执行，先查询副作用或依赖幂等键。

第十二，把 Tool annotation 当成安全事实。 风险等级和审批规则应由本地 Policy 决定。

第十三，把 Server instructions 原样拼进 system prompt。 Server 元数据也可能包含错误或恶意指令，需要信任边界。

第十四，把 MCP Roots 当成沙箱。 Roots 是协议上下文，不会阻止进程读取其它路径。

第十五，在模型调用中途替换工具目录。 目录按轮次冻结，在安全点原子刷新。

第十六，先清理 Worktree，再关闭 Server。 运行资源的释放顺序必须与创建顺序相反。

小结

本文把进程内 Tool Registry 延伸到了 MCP 协议边界：

Agent Harness 是 MCP Host，模型不会直接连接 Server；
Harness 为每条 Server 连接维护 Client Session，并先完成版本与能力协商；
本地 workspace Server 使用 stdio，并绑定当前 Task worktree；
Server command、环境、工作目录和工具 allowlist 都来自 Harness Policy；
tools/list 需要遍历分页、校验 schema 并限制目录规模；
本地稳定别名解决多 Server 名称冲突，Binding 保存真实远端名称；
模型 tool_call_id 与 MCP JSON-RPC request ID 属于不同协议层；
输入在请求前校验，成功的结构化输出按 outputSchema 再校验；
isError 反馈给模型，连接与 contract failure 交给 Recovery；
Hook、审批、Task、Job、Artifact 和 Worktree 继续由 Harness 控制；
Tool Catalog 按模型轮次冻结，在安全点刷新；
关闭 MCP 连接后，才能回收对应的任务工作区。

现在，Agent 使用工具时不再关心能力位于同一进程、Task 子进程还是独立服务。更重要的是，引入协议并没有绕开前面建立的工程边界：模型仍然只能看到 Harness 批准的 schema，每次调用仍然经过权限、审计和恢复流程。

本章使用 stdio 解决了本地受控进程的接入。下一章可以继续实现 远程 MCP：使用 Streamable HTTP 连接共享 Server，并补上认证、Origin 校验、会话恢复、并发限制与多租户凭据隔离。