Agent Harness 工程：接入远程 MCP
Agent Harness 工程：接入远程 MCP
上一章的 MCP Server 跟着 Task 一起启动：
Agent Harness
-> stdio Client
-> Task 子进程
-> Worktree 内的工具
这很适合文件、Shell 和测试工具。每个 Task 有自己的进程，连接生命周期也自然跟随 Worktree。
但 Issue、日历、知识库和数据库网关通常已经是独立服务。它们由团队共享，也不应该为每个 Agent 再启动一份。
于是 Server 搬到了网络另一端：
Agent Harness
-> Streamable HTTP Client
-> Remote MCP Server
-> shared service
代码上，似乎只是把 stdio_client() 换成 streamable_http_client()。
工程上却同时改变了三件事：
- Server 不再跟随 Task 生灭，连接会断，也会重建；
- 每个 HTTP 请求都要重新证明身份和权限；
- 网络中断不代表请求没有执行，更不代表副作用已经取消。
远程 MCP 真正难的从来不是“把 URL 填进去”，而是让身份、会话和业务结果各自待在正确的边界里。
Streamable HTTP 仍然是一套 MCP
按照 MCP Streamable HTTP 规范，Server 对外提供一个统一 MCP endpoint，例如：
https://mcp.example.com/mcp
每条 Client 消息通过 HTTP POST 发出：
POST /mcp
Content-Type: application/json
Accept: application/json, text/event-stream
Server 可以返回：
application/json
一次普通 JSON-RPC 响应
text/event-stream
在 SSE 流中发送进度、通知和最终响应
202 Accepted
已接受 notification 或 response，本次无响应体
可选的 GET /mcp 可以建立 Server 主动消息的 SSE 流。
Client 消息仍由 POST 发送；JSON 与 SSE 只是承载同一套 MCP 消息的不同响应方式。
这里最容易出现三种误解。
SSE 不是另一套工具协议。 不论响应是 JSON 还是 SSE，Client 仍然调用 session.call_tool()。
SSE 断开不等于工具取消。 Client 只是暂时收不到后续事件，Server 端操作可能仍在继续。
Streamable HTTP 不是旧的 HTTP+SSE。 当前规范使用一个同时支持 POST 与 GET 的 MCP endpoint，不再拆成旧式的两个地址。
传输方式改变了，Agent Loop 不应该因此出现一条新的业务分支。
简单共享服务，先保持无状态
不是所有远程 Server 都需要 Session。
一个“根据 ID 查询 Issue”的服务，收到请求后立即返回，最适合无状态模式：
from typing import Annotated
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
mcp = FastMCP(
"issue-query",
stateless_http=True,
json_response=True,
)
class Issue(BaseModel):
id: str
title: str
status: str
ISSUES = {
"ISSUE-42": Issue(
id="ISSUE-42",
title="Add MCP health checks",
status="open",
)
}
@mcp.tool()
async def get_issue(
issue_id: Annotated[
str,
Field(pattern=r"^ISSUE-[0-9]+$"),
],
) -> Issue:
"""Read one issue by its stable ID."""
issue = ISSUES.get(issue_id)
if issue is None:
raise ValueError("issue does not exist")
return issue
if __name__ == "__main__":
mcp.run(transport="streamable-http")
stateless_http=True 表示 Server 不依赖跨请求的 MCP Session；json_response=True 让短请求优先直接返回 JSON。
无状态并不是功能缩水。它更容易放进负载均衡器后面，也更容易横向扩容和逐请求鉴权。
只有确实需要 Server-to-Client notification、长时间进度流、恢复 SSE 或跨请求状态时，才引入 stateful session。先打开状态，再寻找状态存在的理由，只会增加恢复负担。
上面的代码适合本地验证。生产环境仍需要 TLS、认证、Origin 校验、请求大小限制和受控网络出口。
Token、Session 与 tool call ID 各管一件事
远程 MCP 中常见的三个标识不能混用：
Access Token
当前 HTTP 请求代表谁，拥有什么权限
MCP-Session-Id
多个 HTTP 请求属于哪个协议会话
Provider tool_call_id
模型消息中的调用与结果如何闭环
如果 Server 在初始化响应中返回 MCP-Session-Id，Client 会在后续请求中带回它。
但 Session ID 不是登录凭据。
Server 必须对每一个 HTTP 请求验证 Access Token，并确认 Session 仍属于同一用户和租户。只在创建 Session 时鉴权一次，后续仅凭 Session ID 放行，就等于悄悄把 Session 变成了长期凭据。
OAuth 决定访问权限，MCP Session 关联协议状态，Server 访问下游系统时还要使用自己的凭据。
Task checkpoint 也不能用 Session ID 恢复业务进度。Session 失效后可以重建，真正的工作状态仍然保存在 Task、Job 与 Artifact 中。
MCP Server 是 OAuth Resource Server
按照 MCP 授权规范，受保护的 MCP Server 扮演 OAuth Resource Server：
Authorization Server
签发 Access Token
Agent Harness
OAuth Client
MCP Server
验证 Token 的 Resource Server
FastMCP 可以这样装配：
from mcp.server.auth.provider import TokenVerifier
from mcp.server.auth.settings import AuthSettings
from pydantic import AnyHttpUrl
def build_issue_server(
token_verifier: TokenVerifier,
) -> FastMCP:
server = FastMCP(
"issue-service",
stateless_http=True,
json_response=True,
token_verifier=token_verifier,
auth=AuthSettings(
issuer_url=AnyHttpUrl(
"https://auth.example.com"
),
resource_server_url=AnyHttpUrl(
"https://mcp.example.com"
),
required_scopes=["mcp:issues.read"],
),
)
@server.tool()
async def get_issue(
issue_id: str,
) -> Issue:
"""Read one issue by its stable ID."""
issue = ISSUES.get(issue_id)
if issue is None:
raise ValueError(
"issue does not exist"
)
return issue
return server
这里把 TokenVerifier 作为部署依赖传入，而不是演示一个“Token 非空就算通过”的假实现。
真实验证至少包括：
- 签名与允许算法；
- issuer；
- audience 或 resource；
- 过期和生效时间；
- scope；
- 撤销状态。
required_scopes 只是入口最低门槛。工具内部仍要检查当前主体能否读取目标项目，而不是拿到 mcp:issues.read 就能访问所有租户。
如果 MCP Server 还要调用下游 Issue API，它应使用自己的下游凭据，或进行明确的 Token Exchange。不能把 Harness 发来的 Access Token 原样转发，这会形成危险的 token passthrough。
远程地址仍然来自 Harness Policy
模型只选择管理员已经注册的 server_id，不能提供 URL、issuer、scope 和 redirect URI。
from dataclasses import dataclass
@dataclass(frozen=True)
class RemoteMcpServerConfig:
id: str
endpoint: str
resource_url: str
redirect_uri: str
scopes: tuple[str, ...]
allowed_hosts: frozenset[str]
allowed_tools: frozenset[str]
supported_protocol_versions: frozenset[str]
request_timeout_seconds: float = 30.0
stream_read_timeout_seconds: float = 300.0
connect_timeout_seconds: float = 5.0
queue_timeout_seconds: float = 10.0
max_concurrency: int = 8
terminate_on_close: bool = True
endpoint 是 MCP 消息地址：
https://mcp.example.com/mcp
resource_url 是 OAuth 中标识目标 Resource Server 的地址：
https://mcp.example.com
二者可能不同，不能靠删除 /mcp 猜测。
加载配置时要拒绝 URL userinfo、query、fragment、未知 host 和生产环境中的明文 HTTP。开发模式可以显式允许带端口的 loopback HTTP。下面这层校验负责挡住最明显的配置错误：
from ipaddress import ip_address
from urllib.parse import urlsplit
def is_loopback_host(host: str) -> bool:
if host == "localhost":
return True
try:
return ip_address(host).is_loopback
except ValueError:
return False
def validate_registered_url(
value: str,
*,
allowed_hosts: frozenset[str],
development: bool,
) -> None:
parsed = urlsplit(value)
try:
port = parsed.port
except ValueError as exc:
raise ValueError("invalid URL port") from exc
if (
parsed.scheme not in {"https", "http"}
or parsed.hostname is None
or parsed.username is not None
or parsed.password is not None
or parsed.query
or parsed.fragment
):
raise ValueError("invalid registered URL")
if parsed.hostname not in allowed_hosts:
raise ValueError("URL host is not allowed")
if parsed.scheme == "http" and not (
development
and port is not None
and is_loopback_host(parsed.hostname)
):
raise ValueError("plain HTTP is not allowed")
def validate_remote_config(
config: RemoteMcpServerConfig,
*,
development: bool = False,
) -> None:
if not config.allowed_hosts:
raise ValueError("allowed_hosts cannot be empty")
for value in (
config.endpoint,
config.resource_url,
config.redirect_uri,
):
validate_registered_url(
value,
allowed_hosts=config.allowed_hosts,
development=development,
)
timeouts = (
config.request_timeout_seconds,
config.stream_read_timeout_seconds,
config.connect_timeout_seconds,
config.queue_timeout_seconds,
)
if any(value <= 0 for value in timeouts):
raise ValueError("timeouts must be positive")
if (
config.stream_read_timeout_seconds
< config.request_timeout_seconds
):
raise ValueError(
"stream read timeout is too short"
)
if config.max_concurrency < 1:
raise ValueError("max_concurrency must be positive")
if (
not config.scopes
or not config.allowed_tools
or not config.supported_protocol_versions
):
raise ValueError(
"scopes, tools and protocol versions are required"
)
但几行 urlsplit() 只够发现明显错误，不是完整 SSRF 防护。
OAuth discovery 还会访问 Protected Resource Metadata、Authorization Server Metadata、registration 和 token endpoint。攻击者可能让地址指向内网、云 metadata，或在 DNS 解析后切换目标。
生产环境需要：
管理员 allowlist
+ 受控 DNS
+ egress proxy
+ 每次 redirect 后重新校验
+ 最终 IP 范围限制
不要让“模型不能填 URL”成为唯一一道网络边界。
凭据按用户和租户分开存放
共享 Server 不代表共享 Token。
最小凭据分区应该包含：
tenant_id
+ user_id
+ server_id
+ resource_url
= credential partition
import hashlib
from dataclasses import dataclass
from typing import Any, Protocol
@dataclass(frozen=True)
class CredentialPartition:
tenant_id: str
user_id: str
server_id: str
resource_url: str
def storage_prefix(self) -> str:
raw = "\0".join(
(
self.tenant_id,
self.user_id,
self.server_id,
self.resource_url,
)
)
digest = hashlib.sha256(
raw.encode("utf-8")
).hexdigest()
return f"mcp-oauth/{digest}"
class SecretStore(Protocol):
async def read_json(
self,
key: str,
) -> dict[str, Any] | None:
...
async def write_json(
self,
key: str,
value: dict[str, Any],
) -> None:
...
摘要只是不在 key 中暴露用户标识，不是 Token 加密。真正的加密、访问控制、轮换与审计由 Secret Store 负责。
官方 SDK 的 Token Storage 可以接到这层：
from mcp.client.auth import TokenStorage
from mcp.shared.auth import (
OAuthClientInformationFull,
OAuthToken,
)
class SecretBackedTokenStorage(TokenStorage):
def __init__(
self,
secret_store: SecretStore,
partition: CredentialPartition,
) -> None:
prefix = partition.storage_prefix()
self.secret_store = secret_store
self.tokens_key = f"{prefix}/tokens"
self.client_key = f"{prefix}/client"
async def get_tokens(self) -> OAuthToken | None:
payload = await self.secret_store.read_json(
self.tokens_key
)
return (
OAuthToken.model_validate(payload)
if payload is not None
else None
)
async def set_tokens(
self,
tokens: OAuthToken,
) -> None:
await self.secret_store.write_json(
self.tokens_key,
tokens.model_dump(
mode="json",
exclude_none=True,
),
)
async def get_client_info(
self,
) -> OAuthClientInformationFull | None:
payload = await self.secret_store.read_json(
self.client_key
)
return (
OAuthClientInformationFull.model_validate(
payload
)
if payload is not None
else None
)
async def set_client_info(
self,
client_info: OAuthClientInformationFull,
) -> None:
await self.secret_store.write_json(
self.client_key,
client_info.model_dump(
mode="json",
exclude_none=True,
),
)
Token 与动态注册信息不能进入：
- System Prompt；
messages；- Task checkpoint；
- Artifact；
- 普通日志；
- 模型可读的文件系统。
Checkpoint 只保存 credential partition 或 Server 配置 ID，恢复时再由受信任的 Secret Store 读取。
OAuth 交互属于可信 UI，不属于模型
Authorization Code Flow 要把授权地址展示给用户，并接收 callback。
这一步不能由模型完成，也不能把 URL 拼进 Shell 命令打开。
Harness 控制面应该：
- 创建一次性 OAuth transaction；
- 绑定 tenant、user、server 和过期时间；
- 在可信 UI 中跳转授权地址；
- 让 callback 回到固定 redirect URI；
- 校验
state，交回 code； - 使用一次后立即失效。
同一 credential partition 的交互式授权最好串行。若允许同时发起多次，就必须用 transaction ID 区分，不能只拿 partition 当 callback key。
官方 SDK 的 OAuthClientProvider 会处理 metadata discovery、PKCE、state、Token 获取和 refresh。Harness 仍然负责限制 discovery 的网络出口，以及把授权交互留在可信 UI。
远程 Connection 保持与本地相同的外形
上一章的发现与 Adapter 只需要：
connection.config
connection.session
connection.catalog_dirty
远程 Connection 保持这三个属性，就可以复用 Tool Binding、Schema 校验和 Registry。
import asyncio
import hashlib
from collections.abc import Callable
from contextlib import AsyncExitStack
from datetime import timedelta
import httpx
from mcp import ClientSession
from mcp.client.streamable_http import (
streamable_http_client,
)
from mcp.types import (
Implementation,
ServerNotification,
ToolListChangedNotification,
)
class RemoteMcpConnection:
def __init__(
self,
config: RemoteMcpServerConfig,
oauth_provider: httpx.Auth,
) -> None:
self.config = config
self.oauth_provider = oauth_provider
self.catalog_dirty = asyncio.Event()
self.call_slots = asyncio.Semaphore(
config.max_concurrency
)
self._stack: AsyncExitStack | None = None
self._session: ClientSession | None = None
self._get_session_id: (
Callable[[], str | None] | None
) = None
@property
def session(self) -> ClientSession:
if self._session is None:
raise RuntimeError(
"remote MCP connection is not open"
)
return self._session
def session_fingerprint(self) -> str | None:
if self._get_session_id is None:
return None
session_id = self._get_session_id()
if session_id is None:
return None
return hashlib.sha256(
session_id.encode("utf-8")
).hexdigest()[:12]
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,
) -> "RemoteMcpConnection":
if self._stack is not None:
raise RuntimeError(
"remote MCP connection is already open"
)
stack = AsyncExitStack()
self._stack = stack
try:
timeout = httpx.Timeout(
connect=(
self.config.connect_timeout_seconds
),
read=(
self.config
.stream_read_timeout_seconds
),
write=(
self.config.request_timeout_seconds
),
pool=(
self.config.queue_timeout_seconds
),
)
pool_size = (
self.config.max_concurrency + 4
)
http_client = (
await stack.enter_async_context(
httpx.AsyncClient(
auth=self.oauth_provider,
timeout=timeout,
limits=httpx.Limits(
max_connections=pool_size,
max_keepalive_connections=(
pool_size
),
keepalive_expiry=30,
),
follow_redirects=False,
headers={
"User-Agent": (
"agent-harness/0.1.0"
)
},
)
)
)
read_stream, write_stream, get_session_id = (
await stack.enter_async_context(
streamable_http_client(
self.config.endpoint,
http_client=http_client,
terminate_on_close=(
self.config
.terminate_on_close
),
)
)
)
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(
"protocol version is not allowed"
)
if (
initialization.capabilities.tools
is None
):
raise McpContractError(
"server does not advertise tools"
)
self._session = session
self._get_session_id = get_session_id
return self
except BaseException:
self._session = None
self._get_session_id = None
self._stack = None
await stack.aclose()
raise
async def __aexit__(
self,
*exc_info: object,
) -> None:
stack = self._stack
self._session = None
self._get_session_id = None
self._stack = None
if stack is not None:
await stack.aclose()
SDK 负责维护 Accept、MCP-Session-Id、协议版本 Header 和 SSE 的 Last-Event-ID。Harness 不要手写第二套 Session Header 管理。
get_session_id 只用于生成不可逆观测指纹，原始值不能写进日志。
terminate_on_close=True 会在存在 Session 时尝试发送 DELETE；Server 不支持时可以返回 405，Client 仍然正常释放本地资源。
HTTP timeout 也不是一个数字：
connect 控制 DNS、路由与 TLS 建连
write 控制请求体发送
read 允许合法的 SSE 等待
pool 防止无限等待连接池
单次 Tool 调用仍然使用 ClientSession.call_tool(..., read_timeout_seconds=...) 设置业务等待上限。
用 Protocol 表达本地与远程的共同接口
把上一章的具体类型注解改成结构化 Protocol：
from typing import Protocol
class McpToolConfig(Protocol):
id: str
allowed_tools: frozenset[str]
request_timeout_seconds: float
class McpToolConnection(Protocol):
config: McpToolConfig
@property
def session(self) -> ClientSession:
...
然后：
async def discover_tools(
connection: McpToolConnection,
) -> list[McpToolBinding]:
...
McpToolAdapter 的 connection 参数也改成 McpToolConnection，函数体不变。
这样 stdio 与 HTTP 的差异停留在连接层：
initialize
-> tools/list
-> allowlist
-> schema validation
-> catalog snapshot
-> local binding
-> tools/call
模型不需要知道远端返回的是 JSON 还是 SSE。
共享 Server 必须有背压
本地 Task Server 通常只服务一个 Agent Loop，远程 Server 却可能同时服务许多 Task。
HTTP connection pool 只限制 socket 数量，不等于业务并发限制。
class RemoteMcpToolAdapter(McpToolAdapter):
def __init__(
self,
connection: RemoteMcpConnection,
bindings: list[McpToolBinding],
) -> None:
super().__init__(connection, bindings)
self.connection = connection
async def call(
self,
local_name: str,
arguments: dict[str, Any],
) -> str:
try:
async with asyncio.timeout(
self.connection.config
.queue_timeout_seconds
):
await (
self.connection.call_slots.acquire()
)
except TimeoutError:
return (
"Error: remote MCP server is busy"
)
try:
return await super().call(
local_name,
arguments,
)
finally:
self.connection.call_slots.release()
连接与 semaphore 应按 credential partition 维护：
(tenant A, user 1, server X)
-> connection + semaphore
(tenant A, user 2, server X)
-> another connection + semaphore
还要叠加 tenant quota、模型单轮工具数、HTTP pool 和单次调用 timeout。
如果目标 Server 依赖严格顺序，把 max_concurrency 设为 1。不要用无上限 asyncio.gather() 把 Provider 返回的所有 tool call 一次性冲向远端。
“重连”其实是三件完全不同的事
远程 MCP 中最容易误用的词就是“重连”。
SSE 流恢复
Server 为 SSE event 提供 ID 时，Client 可以带 Last-Event-ID 继续读取遗漏消息。
这仍是同一 Session、同一逻辑请求，不会重新发送 tools/call。
MCP Session 重建
如果 Server 返回 404 表示 Session 已终止，Harness 必须：
- 停止向旧连接发送新调用；
- 把在途写操作标记为
unknown； - 完整关闭旧 transport context；
- 创建新 HTTP Client 与
ClientSession； - 重新
initialize； - 重新发现并校验工具；
- 在下一个模型回合原子替换 Catalog。
不能只在旧 ClientSession 上再调用一次 initialize()。Transport 内部仍持有已经失效的 Session ID。
工具重试
工具重试会重新发送一次业务调用。
如果 create_issue 已经执行成功、响应前 Session 失效，新连接再次调用就会创建第二条 Issue。
除非工具携带由 Harness 生成、Server 强制执行的业务幂等键，否则写操作不能自动重放。
三者的区别可以压缩成一句话：
流恢复继续读
会话重建重新握手
工具重试重新做
它们绝不能因为都叫“重连”就共用一个按钮。
断线后的恢复，先承认不知道
不同故障需要不同动作：
本地 Schema 校验失败
交给模型修正
isError=true
交给模型处理，连接继续使用
SSE 短暂中断且可恢复
继续读取，不重放调用
Session 终止
重建连接与 Catalog
401 Unauthorized
refresh 或重新授权，不暴露 Token
403 insufficient_scope
经用户确认后有限 step-up
timeout / 网络中断
写操作进入 unknown，先查询副作用
输出违反 outputSchema
记录 contract failure，不交给模型
unknown 写操作要形成持久化 Recovery Record，而不是只打一行错误日志。Harness 重启后仍然要知道哪些外部结果尚未确认。
即使发送了 MCP cancellation notification，也要由具体工具合同说明外部副作用是否可回滚。请求被取消和现实已经恢复原状，不是同一件事。
Connection 可以复用，生命周期不能混用
把认证、连接、发现和 Registry 放进一个 context：
from contextlib import asynccontextmanager
from dataclasses import replace
from typing import AsyncIterator
@asynccontextmanager
async def remote_task_state(
state: HarnessState,
config: RemoteMcpServerConfig,
partition: CredentialPartition,
oauth_provider: httpx.Auth,
) -> AsyncIterator[HarnessState]:
validate_remote_config(config)
if config.id != partition.server_id:
raise ValueError(
"partition does not match server"
)
if (
config.resource_url
!= partition.resource_url
):
raise ValueError(
"partition does not match resource"
)
async with RemoteMcpConnection(
config,
oauth_provider,
) as connection:
bindings = await discover_tools(connection)
adapter = RemoteMcpToolAdapter(
connection,
bindings,
)
task_registry = state.tool_registry.fork()
install_mcp_tools(
task_registry,
adapter,
)
yield replace(
state,
tool_registry=task_registry,
)
工具在认证、初始化和目录校验成功之前不会暴露给模型。派生出的 HarnessState 也只能在这个 context 内使用；退出后连接已经关闭，捕获它的 MCP handler 与 Task Registry 必须一起丢弃。
实际系统可以由 Connection Manager 复用同一 credential partition 的健康连接，但要使用引用计数或 lease。一个 Task 退出时，不能关闭另一个 Task 正在使用的连接，更不能为了省一次 TLS 握手跨用户复用。
Origin、CORS 与 SSRF 保护不同方向
这三个词经常一起出现，却不是同一层安全措施。
Origin 保护 MCP Server
Streamable HTTP Server 必须检查请求中的 Origin。存在且不在 allowlist 中时，应返回 403，降低 DNS rebinding 攻击风险。
后端 httpx Client 通常没有 Origin，不需要伪造。Server 要区分“没有 Origin 的非浏览器 Client”和“带 Origin 的浏览器请求”。
CORS 控制浏览器能否读取响应
浏览器直连时，要精确配置 allowed origins、methods 和 headers。若浏览器需要读取 Session Header，还要暴露 MCP-Session-Id。
CORS 不是认证。Server 仍然要验证每次请求的 Token。
SSRF 保护 Agent Harness
Harness 会主动访问 MCP endpoint、OAuth metadata 与 token endpoint，因此自己也可能成为 SSRF 发起者。
防护要覆盖每个 discovery URL、DNS 最终地址和 HTTP redirect。最可靠的组合是配置 allowlist 加受控 egress，而不是相信远端返回的下一条 URL。
观测连接，但不要把凭据写进日志
一次远程调用可以记录：
trace_id
task_id
server_id
tenant_fingerprint
session_fingerprint
catalog_snapshot_id
provider_tool_call_id
remote_tool_name
connection_generation
queue_wait_ms
duration_ms
result_class
不能记录：
Authorization header
access token
refresh token
authorization code
raw MCP-Session-Id
OAuth client secret
未脱敏的工具参数和结果
Session fingerprint 只用于关联日志，不能用来恢复连接。
还要检查 HTTP Client 与 MCP SDK 的 debug 日志，避免底层库打印 Header、URL query 和原始 Session ID。
小结
这一章把本地 MCP Client 延伸到了远程共享服务：
- Streamable HTTP 使用统一 MCP endpoint；
- JSON 与 SSE 承载同一套协议消息；
- 简单 Server 优先使用 stateless HTTP；
- Access Token、MCP Session 与 Provider tool call ID 各自分层；
- MCP Server 作为 OAuth Resource Server 逐请求鉴权；
- endpoint、resource、scope 和 allowlist 来自 Harness Policy；
- Token 按 tenant、user、server 和 resource 分区；
- OAuth 交互留在可信 UI，凭据只进 Secret Store；
- 远程 Connection 复用初始化、发现、Binding 和 Schema 校验；
- semaphore、连接池和 tenant quota 共同提供背压；
- 流恢复、会话重建与工具重试严格分开；
- 断线后的写操作进入
unknown，不盲目重放； - Origin、CORS 和 SSRF 各自保护不同方向；
- Trace 使用指纹关联，不记录 Token 与原始 Session。
现在，Agent 可以用同一套 Tool Registry 连接 Task 内的本地 Server，也可以连接团队共享的远程 Server。
传输跨过了网络，前面建立的权限、审批、Task、Job、Worktree 和恢复边界却没有被绕开。
下一章，也是这个系列的最后一章，会把这些模块重新放回同一张图中：从最初的 Agent Loop 出发，看看我们究竟怎样一步步搭出了一套真正能把事情做完的 Agent Harness。