MCP tools

Use MCP to connect tools supplied by another process or service. Harness composes Pydantic AI's MCP Capability instead of adding another transport client.

A Harness SDK AgentSpec does not accept Harness UI's top-level mcp_servers resource field. The SDK uses Capabilities; the application owns resource IDs and configuration files.

Native MCPCopy Anchor Link

MCP uses Pydantic AI's native MCP Capability. Keep it in AgentSpec.capabilities; Harness does not define a second MCP client, protocol, server schema, or peer mcp_servers field.

A URL server with local execution (local: True) can be reconstructed directly from an AgentSpec document:

from a13n_harness import AgentSpec

agent_spec = AgentSpec.from_dict(
    {
        "capabilities": [
            {
                "MCP": {
                    "url": "https://mcp.example.com/mcp",
                    "id": "knowledge",
                    "local": True,
                    "native": False,
                    "allowed_tools": ["search"],
                }
            }
        ]
    }
)

The default a13n-harness installation includes Pydantic AI's MCP client runtime, so local execution over URL and stdio transports needs no separate Harness extra. For process-local inputs such as an in-process server, transport, script path, or prebuilt MCPToolset, construct pydantic_ai.capabilities.MCP in trusted code and pass it to HarnessBuilder().build(..., capabilities=...). A Host can attach a fresh upstream MCP projection to RunBindings.capabilities while owning the entered client's lifetime separately. Do not reuse mutable Run projections or share authenticated clients across different authority/header bindings. defer_loading=True uses upstream load_capability under the same Harness tool boundaries. Use native=True, local=False when the selected model provider should execute a URL MCP server natively.

Host-owned clientsCopy Anchor Link

When several Runs should use the same server state, trusted Host code can own an entered FastMCP client and create a new upstream projection for each Run:

from fastmcp import Client
from pydantic_ai.capabilities import MCP
from pydantic_ai.mcp import MCPToolset

from a13n_harness import RunBindings


async def use_host_client(executable):
    async with Client("https://mcp.example.com/mcp", mode="auto") as client:
        results = []
        for prompt in ("Create a workspace", "Inspect that workspace"):
            projection = MCPToolset(client, id="workspace", cache_tools=False)
            bindings = RunBindings.embedded(
                capabilities=(MCP(id="workspace", local=projection),)
            )
            results.append(await executable.run(prompt, bindings=bindings))
        return results

Configure authentication and any input handlers on the Host client before entering it. The Host owns shutdown, current authorization, callback routing, and isolation of exact bindings; Harness does not pool connections or persist clients in continuation state. auto delegates modern discovery and legacy negotiation to the FastMCP client. Explicit legacy and 2026-07-28 modes are available on the code-first client. The FastMCP client owns multi-round input and request-state handling, not another Harness Agent loop. A disconnected client is not permission to replay an uncertain business call.

Run-scoped headers with ContextualMCPCopy Anchor Link

Use ContextualMCP when a URL-based MCP server needs headers derived from the current logical Harness Run. The ContextualMCP definition stores an inert URL recipe. When Pydantic AI binds Capabilities for a Run, ContextualMCP resolves the headers and constructs a fresh upstream MCP before native tools or a local MCP Toolset are extracted.

For common identity, lineage, Run, and metadata values, use the declarative resolver:

from a13n_harness import (
    AgentIdentityRef,
    HarnessBuilder,
    RunBindings,
)
from a13n_harness.mcp import (
    ContextualMCP,
    MCPContextHeaderBinding,
    MCPContextHeaders,
    MCPContextHeadersConfig,
)

mcp = ContextualMCP(
    "https://mcp.example.com/mcp",
    id="knowledge",
    native=True,
    local=None,
    headers={"X-Application": "support"},
    headers_factory=MCPContextHeaders(
        MCPContextHeadersConfig(
            headers={
                "X-Run-ID": MCPContextHeaderBinding("context.run_id"),
                "X-Thread-ID": MCPContextHeaderBinding("context.thread_id"),
                "X-User-ID": MCPContextHeaderBinding("identity.user_id"),
                "X-Request-Context": MCPContextHeaderBinding(
                    "context.metadata.request_context",
                    required=False,
                ),
            }
        )
    ),
)

executable = HarnessBuilder().build(
    agent_spec,
    output_type=str,
    model=model,
    capabilities=(mcp,),
)

bindings = RunBindings.embedded(
    identity=AgentIdentityRef(
        issuer="my-host",
        subject="support-agent",
        user_id="user-123",
        agent_id="agent-support",
    ),
    metadata={
        "request_context": {
            "region": "us-east",
            "labels": ["interactive", "priority"],
        }
    },
)
result = await executable.run("Find the account record", bindings=bindings)

RunBindings.metadata is the intended place for additional per-Run JSON values. Put an exact top-level key there, then select it through context.metadata.<key>. Do not attach ad hoc attributes to AgentContext or encode a nested reflection path.

The declarative resolver supports these exact sources:

A selected string is sent unchanged. JSON numbers, booleans, objects, and arrays use finite, sorted-key, compact JSON. For example, {"region": "us-east", "labels": ["interactive"]} becomes {"labels":["interactive"],"region":"us-east"}. A missing value or None fails a required binding and omits an optional binding.

Header names from headers= and the resolved factory result must not overlap case-insensitively. authorization_token, allowed_tools, description, and defer_loading retain upstream MCP behavior.

Custom header factoriesCopy Anchor Link

Use a custom synchronous or asynchronous factory when the declarative resolver's sources are not enough. It receives the complete trusted AgentContext for the logical Run and returns an exact string-to-string mapping:

from collections.abc import Mapping

from a13n_harness import AgentContext
from a13n_harness.mcp import ContextualMCP


def resolve_mcp_headers(context: AgentContext) -> Mapping[str, str]:
    return {
        "X-Run-ID": context.run_id,
        "X-Agent-Instance-ID": context.instance.agent_instance_id,
        "X-Tenant-ID": context.instance.identity.require_claim("tenant_id"),
    }


mcp = ContextualMCP(
    "https://mcp.example.com/mcp",
    id="tenant-tools",
    headers_factory=resolve_mcp_headers,
    native=True,
    local=None,
)

An async factory has the same input and output contract:

async def resolve_mcp_headers(context: AgentContext) -> Mapping[str, str]:
    route = await route_store.resolve(context.instance.identity)
    return {"X-Route": route}

The factory runs once per logical Harness Run. Internal model-recovery attempts reuse the same active upstream MCP and header snapshot; another logical Run resolves a fresh snapshot. The factory is trusted Host code, so it may read current Run services deliberately, but model content cannot choose sources or call the factory directly.

Local and provider-native executionCopy Anchor Link

ContextualMCP accepts URL-based upstream execution only:

Prebuilt clients, transports, in-process servers, scripts, and prebuilt Toolsets already own their connection setup. Use native MCP directly for those values rather than combining them with ContextualMCP.

The URL is explicit trusted configuration. Harness requires an HTTP(S) URL for ContextualMCP. Upstream MCP integrations validate the URL, transport, authorization, and provider. Harness does not guess whether URL components contain credentials.

Host-authored configurationCopy Anchor Link

A Host can expose the same URL-based path through its own trusted configuration model. Preserve the ContextualMCP fields and exact execution selection rather than inventing a second MCP runtime. Persist only credential-free desired configuration; resolve headers, short-lived credentials, and current routing through process-local factories when constructing the Capability.

An Agent can select multiple MCP servers when each has a unique id. Use code-first ContextualMCP when configuration requires callable factories, current identity, static headers, or an out-of-band secret resolver.

Group tools from large local MCP serversCopy Anchor Link

Pass a local MCP or ContextualMCP Capability as a ToolProxyGroup source inside ToolProxyCapability(groups=...) to expose grouped discovery instead of every tool schema. Select native=False, local=True; provider-native tools and deferred-loading sources are not proxy targets. Native composition preserves fresh Run binding and contextual headers, and calls still use the original MCP Toolset and transport. It reduces model context, not MCP initialization or tool-listing work.

Result boundaryCopy Anchor Link

Locally executed MCP tools are ordinary dynamically discovered function tools. Their text and JSON returns cross the mandatory Harness result boundary and default to explicit truncation rather than spill when oversized. This bounds the value integrated into model history; it does not impose a transport-body or process-memory limit before the MCP client receives the result. Provider-native MCP execution remains on the provider path and does not cross the local function-tool boundary.