load_mcp

Python · package toolnexus · SPEC §2 · python/src/toolnexus/mcp_source.py

async def load_mcp(
    input: str | dict[str, Any],
    wait_for: Any = None,
) -> McpSource

Parses config with parse_mcp_config, then connects to every enabled server — local stdio via a child process, remote via streamable-HTTP (falling back to SSE) — lists each server’s tools, and converts them into a flat list[Tool], name-prefixed by server (sanitize(server) + "_" + sanitize(tool)). One bad server never sinks the load: a failed connect is isolated to that server’s status entry.

When to use it

You have an mcp.json (or equivalent dict) and want every enabled server’s tools as Tools, ready to hand to an adapter or the client loop, in one call.
You are wiring MCP servers without skills, builtins, or a toolkit — this is the bare MCP source.
You need per-server status ("connected" | "disabled" | "failed") to decide whether to warn, retry, or just proceed with whatever connected.

Why this and not the alternative

For a hung or slow server you want to bound and cancel, use load_mcp_with_context — load_mcp delegates to it with no cancellation token, so a healthy config behaves identically either way. For the tool inventory without anything staying connected afterward, use list_mcp_tools instead.

Examples

1. Every server disabled — nothing connects, nothing hangs

examples/mcp.json is the fixture every port is tested against. Flip both servers off before loading it, and load_mcp returns immediately with an empty tool list and a status entry per server.

import asyncio
from toolnexus import load_mcp, parse_mcp_config

config = parse_mcp_config("examples/mcp.json")
for cfg in config.values():
    cfg["enabled"] = False


async def main():
    source = await load_mcp(config)
    try:
        assert source.tools == []
        assert source.status == {"everything": "disabled", "example-remote": "disabled"}
    finally:
        # close() is safe even when nothing connected.
        await source.close()

    print("ok:", source.status)


asyncio.run(main())

2. A bad local server fails fast and stays isolated

A command that does not exist raises immediately — no 30-second timeout to wait out. The good server still connects (here, “connects” by virtue of being disabled and thus never dialed).

import asyncio
from toolnexus import load_mcp

config = {
    "broken": {
        "type": "local",
        "command": ["/no/such/binary-toolnexus-docs-example"],
    },
    "off": {
        "type": "local",
        "command": ["echo", "hi"],
        "enabled": False,
    },
}


async def main():
    source = await load_mcp(config)
    try:
        # The bad server is isolated as "failed" — the call itself does not raise.
        assert source.status["broken"] == "failed"
        assert source.status["off"] == "disabled"
        assert source.tools == []
    finally:
        await source.close()

    print("ok:", source.status)


asyncio.run(main())

3. Validate with parse_mcp_config, then decide what to load

The realistic shape: parse the config yourself first, inspect it, and only call load_mcp on the subset you actually want connected — here, none of it, which is exactly the CI-safe path for a config that names real network services.

import asyncio
from toolnexus import load_mcp, parse_mcp_config

config = parse_mcp_config("examples/mcp.json")
assert sorted(config.keys()) == ["everything", "example-remote"]

# example-remote already ships disabled; disable "everything" too so this call
# never tries to spawn a real process.
config["everything"]["enabled"] = False


async def main():
    source = await load_mcp(config)
    try:
        assert source.tools == []
        assert all(status == "disabled" for status in source.status.values())
    finally:
        await source.close()

    print("ok:", sorted(source.status))


asyncio.run(main())

Options

Parameter	Type	What it does
`input`	`str \| dict`	A path or dict, in any of the forms `parse_mcp_config` accepts.
`wait_for`	`Callable \| None`	The §10 suspension resolver. Registered as the MCP elicitation callback when present — see `elicitation_to_request`. `None` means a server that elicits gets no answer.

Returns — `McpSource`

Member	Type	What it is
`tools`	`list[Tool]`	Every tool from every connected server, name-prefixed `<server>_<tool>`.
`status`	`dict[str, McpStatus]`	Per-server `"connected"`, `"disabled"`, or `"failed"`.
`close()`	`async () -> None`	Tears down every connection this call opened. Always call it (or let a context manager, once you have wired one, do it for you).