Server specs + registry

MCPServerSpec.stdio(...)

Launches an MCP server as a subprocess and talks to it over stdio.

from loomflow.mcp import MCPServerSpec
 
spec = MCPServerSpec.stdio(
    name="git",                            # registry id
    command="uvx",                         # the binary to launch
    args=["mcp-server-git", "--repo", "/Users/me/code/myrepo"],
    env={"GIT_AUTHOR_NAME": "agent"},      # optional extra env
    cwd="/Users/me/code/myrepo",           # optional cwd
)

The name becomes the prefix for tool-name disambiguation (git.commit etc.).

MCPServerSpec.http(...)

Talks to a hosted MCP server over HTTP / Server-Sent Events.

spec = MCPServerSpec.http(
    name="hosted",
    url="https://example.com/mcp/",
    headers={
        "Authorization": "Bearer sk-...",
        "X-Tenant-Id": "acme",
    },
    timeout_s=30.0,                        # optional
)

The framework keeps a single SSE connection open per server for the lifetime of the registry; reconnects on transient failures.

MCPRegistry. Combine multiple servers

from loomflow import Agent
from loomflow.mcp import MCPRegistry, MCPServerSpec
 
registry = MCPRegistry([
    MCPServerSpec.stdio("git", "uvx", ["mcp-server-git", "--repo", "."]),
    MCPServerSpec.stdio("fs",  "uvx", ["mcp-server-filesystem", "--root", "."]),
    MCPServerSpec.http("hosted", url="https://example.com/mcp/"),
])
 
agent = Agent("...", model="claude-opus-4-7", tools=registry)

MCPRegistry implements the ToolHost protocol. Pass it directly to Agent(tools=...).

Tool-name conflict resolution

If two servers expose a tool with the same simple name:

• Both get qualified, git.status and fs.status.
• The agent sees both qualified names in the tool list.
• Either qualified or bare form is accepted at call time; the registry strips the prefix before forwarding.

If only one server has the name, it stays bare:

git.commit                # only git ships commit
status, fs.status, git.status     # both ship status; both qualified

Listing tools at runtime

tools = await registry.list_tools()
for t in tools:
    print(t.name, t.description)

list_tools is a coroutine. The registry queries each server’s tools/list endpoint on first call and caches the result.

Lazy initialization

MCPRegistry connects to its servers lazily. The subprocess launches and the HTTP connection opens on the first agent.run. This keeps Agent(...) synchronous and avoids paying for unused servers in tests.

Cleanup

For long-lived processes, hold the registry across runs and let Python’s garbage collector clean up. For short-lived processes (scripts), the framework’s atexit handler closes stdio subprocesses and HTTP connections cleanly.

Errors

When a server is unreachable for an entire agent.run, the run returns RunResult(interrupted=True, interruption_reason="mcp:<name>").