Agent Harness
    -> stdio Client
    -> Task 子进程
    -> Worktree 内的工具

Agent Harness
    -> Streamable HTTP Client
    -> Remote MCP Server
    -> shared service

POST /mcp
Content-Type: application/json
Accept: application/json, text/event-stream

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")

from mcp.server.auth.provider import TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.fastmcp import FastMCP
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

from dataclasses import dataclass


@dataclass(frozen=True)
class RemoteMcpServerConfig:
    id: str
    endpoint: str
    resource_url: str
    redirect_uri: str
    scopes: tuple[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

https://mcp.example.com/mcp

https://mcp.example.com

import ipaddress
from urllib.parse import urlsplit


def validate_remote_url(
    value: str,
    *,
    allow_loopback_http: bool = False,
) -> None:
    parsed = urlsplit(value)
    if parsed.username or parsed.password:
        raise ValueError("URL userinfo is not allowed")
    if parsed.query or parsed.fragment:
        raise ValueError(
            "query and fragment are not allowed"
        )
    if parsed.hostname is None:
        raise ValueError("URL must contain a host")

    try:
        port = parsed.port
    except ValueError as exc:
        raise ValueError("URL contains an invalid port") from exc

    if parsed.scheme == "https":
        return
    if not allow_loopback_http or parsed.scheme != "http":
        raise ValueError("remote MCP URLs must use HTTPS")

    try:
        address = ipaddress.ip_address(parsed.hostname)
        is_loopback = address.is_loopback
    except ValueError:
        is_loopback = parsed.hostname == "localhost"
    if not is_loopback:
        raise ValueError(
            "plain HTTP is only allowed on loopback"
        )
    if port is None:
        raise ValueError(
            "loopback HTTP must use an explicit port"
        )


def validate_remote_config(
    config: RemoteMcpServerConfig,
    *,
    development: bool = False,
) -> None:
    validate_remote_url(
        config.endpoint,
        allow_loopback_http=development,
    )
    validate_remote_url(
        config.resource_url,
        allow_loopback_http=development,
    )
    validate_remote_url(
        config.redirect_uri,
        allow_loopback_http=development,
    )
    if config.max_concurrency < 1:
        raise ValueError("max_concurrency must be positive")
    if not config.allowed_tools:
        raise ValueError("allowed_tools cannot be empty")
    if not config.scopes or any(
        not scope.strip() for scope in config.scopes
    ):
        raise ValueError("scopes cannot be empty")

    timeout_values = (
        config.request_timeout_seconds,
        config.stream_read_timeout_seconds,
        config.connect_timeout_seconds,
        config.queue_timeout_seconds,
    )
    if any(value <= 0 for value in timeout_values):
        raise ValueError("timeouts must be positive")
    if (
        config.stream_read_timeout_seconds
        < config.request_timeout_seconds
    ):
        raise ValueError(
            "stream read timeout cannot be shorter "
            "than request timeout"
        )

config = state.remote_mcp_servers[server_id]

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:
        ...

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
        )
        if payload is None:
            return None
        return OAuthToken.model_validate(payload)

    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
        )
        if payload is None:
            return None
        return OAuthClientInformationFull.model_validate(
            payload
        )

    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,
            ),
        )

from typing import Protocol


class OAuthInteraction(Protocol):
    async def present_authorization_url(
        self,
        partition: CredentialPartition,
        authorization_url: str,
    ) -> None:
        ...

    async def wait_for_callback(
        self,
        partition: CredentialPartition,
    ) -> tuple[str, str | None]:
        ...

from mcp.client.auth import OAuthClientProvider
from mcp.shared.auth import OAuthClientMetadata
from pydantic import AnyUrl


def build_oauth_provider(
    config: RemoteMcpServerConfig,
    partition: CredentialPartition,
    storage: SecretBackedTokenStorage,
    interaction: OAuthInteraction,
) -> OAuthClientProvider:
    async def redirect_handler(
        authorization_url: str,
    ) -> None:
        await interaction.present_authorization_url(
            partition,
            authorization_url,
        )

    async def callback_handler() -> tuple[str, str | None]:
        return await interaction.wait_for_callback(
            partition
        )

    return OAuthClientProvider(
        server_url=config.resource_url,
        client_metadata=OAuthClientMetadata(
            client_name="Agent Harness",
            redirect_uris=[
                AnyUrl(config.redirect_uri),
            ],
            grant_types=[
                "authorization_code",
                "refresh_token",
            ],
            response_types=["code"],
            token_endpoint_auth_method="none",
            scope=" ".join(config.scopes),
        ),
        storage=storage,
        redirect_handler=redirect_handler,
        callback_handler=callback_handler,
    )

connection.config
connection.session
connection.catalog_dirty

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: OAuthClientProvider,
    ) -> 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,
            )
            max_connections = (
                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=max_connections,
                            max_keepalive_connections=(
                                max_connections
                            ),
                            keepalive_expiry=30.0,
                        ),
                        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(
                    "negotiated 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()

connection.config.id
connection.config.allowed_tools
connection.config.request_timeout_seconds
connection.session

from typing import Protocol


class McpToolConfig(Protocol):
    @property
    def id(self) -> str:
        ...

    @property
    def allowed_tools(self) -> frozenset[str]:
        ...

    @property
    def request_timeout_seconds(self) -> float:
        ...


class McpToolConnection(Protocol):
    @property
    def config(self) -> McpToolConfig:
        ...

    @property
    def session(self) -> ClientSession:
        ...

async def discover_tools(
    connection: McpToolConnection,
) -> list[McpToolBinding]:
    # 函数体与上一章保持一致
    ...

from typing import Any


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; "
                "retry after other calls finish"
            )

        try:
            return await super().call(
                local_name,
                arguments,
            )
        finally:
            self.connection.call_slots.release()

(tenant A, user 1, server X) -> connection + semaphore
(tenant A, user 2, server X) -> connection + semaphore
(tenant B, user 1, server X) -> connection + semaphore

from contextlib import asynccontextmanager
from dataclasses import replace
from typing import AsyncIterator


@asynccontextmanager
async def remote_task_state(
    state: HarnessState,
    config: RemoteMcpServerConfig,
    partition: CredentialPartition,
    secret_store: SecretStore,
    interaction: OAuthInteraction,
) -> AsyncIterator[HarnessState]:
    validate_remote_config(config)
    if config.id != partition.server_id:
        raise ValueError(
            "credential partition does not match server"
        )
    if config.resource_url != partition.resource_url:
        raise ValueError(
            "credential partition does not match resource"
        )

    storage = SecretBackedTokenStorage(
        secret_store,
        partition,
    )
    oauth_provider = build_oauth_provider(
        config,
        partition,
        storage,
        interaction,
    )

    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,
        )

async def run_remote_agent(
    state: HarnessState,
    config: RemoteMcpServerConfig,
    partition: CredentialPartition,
    secret_store: SecretStore,
    interaction: OAuthInteraction,
    user_message: str,
) -> str:
    async with remote_task_state(
        state,
        config,
        partition,
        secret_store,
        interaction,
    ) as task_state:
        return await run_agent_loop(
            task_state,
            user_message,
        )

MCP-Session-Id

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
完整的敏感工具参数与结果