load_mcp_with_context

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

async def load_mcp_with_context(
    input: str | dict[str, Any],
    wait_for: Any = None,
    *,
    cancel: asyncio.Event | None = None,
) -> McpSource

The same connect-and-convert as load_mcp — in fact load_mcp just calls this with cancel=None — plus two bounds load_mcp does not surface on its own: every server’s connect + list is wrapped in asyncio.timeout(server_timeout), so a hung endpoint marks only that server "failed" instead of blocking the whole load, and an optional cancel event lets a parent abort the entire load — tearing down whatever connected so far — instead of waiting it out server by server.

When to use it

A server might hang mid-handshake (a stuck subprocess, a streamable-HTTP endpoint that never answers) and you need the load to keep moving instead of stalling on it.
You are wiring MCP load into something with its own cancellation — a request timeout, a user hitting “stop”, a parent asyncio task being cancelled — and want that to abort cleanly instead of leaking a child process.

Why this and not the alternative

Examples

1. A hung server is bounded by its own timeout, not the sleep

The child process here never writes a byte of MCP protocol — it just sleeps for 5 seconds. The server’s configured timeout is 300ms, so load_mcp_with_context gives up on it in well under a second and marks it "failed", instead of the caller waiting out the full sleep.

import asyncio
import sys
import time
from toolnexus import load_mcp_with_context

config = {
    "hung": {
        "type": "local",
        "command": [sys.executable, "-c", "import time; time.sleep(5)"],
        "timeout": 300,  # ms — far shorter than the 5s sleep
    },
}


async def main():
    started = time.monotonic()
    source = await load_mcp_with_context(config)
    elapsed = time.monotonic() - started
    try:
        assert source.status["hung"] == "failed"
        # Bounded by the server's own timeout, not the 5-second sleep.
        assert elapsed < 3.0
    finally:
        await source.close()

    print("ok: failed after", round(elapsed, 2), "s")


asyncio.run(main())

2. A pre-set cancel event aborts the whole load

Set the event before calling, and nothing connects at all — the whole load raises asyncio.CancelledError instead of returning a partial McpSource.

import asyncio
from toolnexus import load_mcp_with_context

config = {
    "off": {"type": "local", "command": ["echo", "hi"], "enabled": False},
    # Enabled, so the loop reaches the cancel check for it — and never gets to dial it.
    "would-connect": {"type": "local", "command": ["echo", "hi"]},
}


async def main():
    cancel = asyncio.Event()
    cancel.set()  # already fired before the call starts

    try:
        await load_mcp_with_context(config, cancel=cancel)
        raise AssertionError("expected CancelledError")
    except asyncio.CancelledError:
        pass

    print("ok: cancelled before anything connected")


asyncio.run(main())

3. Disabled, broken, and hung servers together — one status map

Three servers, three fates, one call: off never dials, broken fails fast on a missing binary, hung times out on its own per-server budget. No cancel event is set, so the load runs to completion and reports all three.

import asyncio
import sys
from toolnexus import load_mcp_with_context

config = {
    "off": {"type": "local", "command": ["echo", "hi"], "enabled": False},
    "broken": {"type": "local", "command": ["/no/such/binary-toolnexus-docs-example"]},
    "hung": {
        "type": "local",
        "command": [sys.executable, "-c", "import time; time.sleep(5)"],
        "timeout": 300,
    },
}


async def main():
    source = await load_mcp_with_context(config, cancel=asyncio.Event())
    try:
        assert source.status == {
            "off": "disabled",
            "broken": "failed",
            "hung": "failed",
        }
        assert source.tools == []
    finally:
        await source.close()

    print("ok:", source.status)


asyncio.run(main())

Options

Parameter	Type	What it does
`input`	`str \| dict`	Same accepted shapes as `load_mcp` — see `parse_mcp_config`.
`wait_for`	`Callable \| None`	The §10 suspension resolver, wired as the elicitation callback when present.
`cancel`	`asyncio.Event \| None`	When set (before or during the call), aborts the whole load: releases every already-connected server and raises `asyncio.CancelledError`. Distinct from a per-server timeout, which isolates only that server.

Returns

Same McpSource as load_mcp — tools, status, close().