MCP Client

The Model Context Protocol (MCP) lets an agent call tools that live in an external server — a CRM, a knowledge base, a ticketing system. NeoGraph is an MCP client: you bind those tools to nodes and the LLM calls them. The interesting part is everything you get for free once they are neograph nodes — per-node tool binding, per-run identity, and a checkpointed human gate before any mutation fires.

This walkthrough follows examples/23_mcp_client_selective_binding.py, a CRM deal-review pipeline talking to a real MCP server (a plain FastMCP server spawned as a stdio subprocess). The MCP layer is genuine end-to-end — tool discovery, per-operator auth, structured results all round-trip over real stdio JSON-RPC. The only fakes are the LLMs, so the example runs with no API keys and no network beyond the local subprocess.

The battery: one call, a factory per tool

The shipped neograph[mcp] battery turns a typed server config into the {name: async factory} dict that compile(tool_factories=...) already accepts:

from neograph_mcp import StdioServer, mcp_tool_factories

CRM = StdioServer(command=sys.executable, args=[DEMO_SERVER])

def token_provider(configurable: dict) -> str:
    return configurable.get("mcp_auth", "anon")

FACTORIES = mcp_tool_factories(
    {"crm": CRM},
    token_provider=token_provider,
    namespace=False,  # single server → bare tool names
    # Opt each reader into typed results — see Beat 5.
    output_models={"crm_search": CrmSearchResult, "kb_lookup": KbResult},
)
# FACTORIES == {"crm_search": <factory>, "kb_lookup": <factory>,
#               "get_deal": <factory>, "update_deal": <factory>, ...}

mcp_tool_factories connects to the server once to enumerate its tools, then returns a dict — one async factory per tool. It is an overridable default: the seam is compile(tool_factories=...), and the battery is one convenient way to fill it. When you want house rules, hand-roll the factory (see The escape hatch below). The returned dict being a dict — not one bundled agent — is what makes the next beat possible.

Beat 1 — selective binding

The point of an MCP client over “one agent holds every tool” is least privilege per node. Slice the battery’s dict so each node binds only the tools it needs:

READER_TOOLS = ["crm_search", "kb_lookup"]  # read-only slice
MUTATOR_TOOLS = ["update_deal"]             # the mutating slice

@node(
    mode="agent",
    outputs={"result": ResearchBrief, "tool_log": list[ToolInteraction]},
    model="research",
    prompt="research/brief",
    # This agent CANNOT call update_deal — it is not in its tools=.
    tools=[
        Tool("crm_search", budget=2, idempotent=True),
        Tool("kb_lookup", budget=1, idempotent=True),
    ],
    llm_config={"announce_tool_budget": True},
)
def research() -> ResearchBrief: ...

@node(
    mode="act",  # act mode = mutating tools
    outputs=ActionOutcome,
    model="action",
    prompt="action/apply",
    tools=[Tool("update_deal", budget=1)],  # the mutating tool ONLY
    gate_tools_when=lambda state: {"pending_tool": "update_deal", "reason": "Approve CRM mutation?"},
)
def apply_action() -> ActionOutcome: ...

Each node’s compile() gets only its slice, so the least-privilege boundary is enforced at two levels:

research_graph = compile(
    research_pipeline, llm_factory=llm_factory, prompt_compiler=prompt_compiler,
    tool_factories={k: FACTORIES[k] for k in READER_TOOLS},   # readers only
)
action_graph = compile(
    action_pipeline, llm_factory=llm_factory, prompt_compiler=prompt_compiler,
    tool_factories={k: FACTORIES[k] for k in MUTATOR_TOOLS},  # the mutator only
    checkpointer=MemorySaver(),
)

The research agent literally cannot mutate — update_deal is neither in its tools= nor in its tool_factories. One client, many nodes, least privilege per node.

Beat 2 — raw BaseTool passthrough and the lint guard

A battery factory yields a real LangChain BaseTool; you can hand one straight to a node in tools=[raw_tool] with zero ceremony:

raw_search = await FACTORIES["crm_search"]({"configurable": {"mcp_auth": "svc"}}, None)

@node(mode="agent", outputs=ResearchBrief, model="research",
      prompt="research/brief", tools=[raw_search])
def raw_research() -> ResearchBrief: ...

MCP tools are async-only — langchain-mcp-adapters produces a StructuredTool backed by a coroutine with no sync func, so .invoke() raises. lint() catches this before you run:

issues = [i for i in lint(raw_pipeline) if i.kind == "tool_requires_async_driver"]
# [Issue(kind="tool_requires_async_driver",
#        message="Node 'raw-research': tool 'crm_search' is async-only … Drive with arun()")]

So a sync run() fails loud with a ConfigurationError pointing you to arun(); the async driver runs the tool loop fine. The guard is compile-time when the tool is a raw BaseTool (lint can introspect it) and runtime for the factory path.

Beat 3 — per-run identity

Each run carries its own identity. The token_provider reads config['configurable']['mcp_auth'] and mints the token; neograph never inspects it — it only carries it from config to the tool.

for operator in ("operator-A", "operator-B"):
    result = await arun(
        research_graph,
        input={"node_id": "REVIEW-1"},
        config={"configurable": {"mcp_auth": operator}},
    )
    tool_log = result["research_tool_log"]
    who = {i.tool_name: i.typed_result.acting_as for i in tool_log}
    # operator-A → {'crm_search': 'operator-A', 'kb_lookup': 'operator-A'}
    # operator-B → {'crm_search': 'operator-B', 'kb_lookup': 'operator-B'}

The server echoes the identity it saw under acting_as, so each run’s tools observably carried that operator’s token — read straight back out of the tool log.

stdio has no HTTP headers. Over stdio the token rides as a tool argument (the demo server declares a token arg and echoes it); the battery injects the run’s token into that argument, overriding any model-supplied value so identity is framework-carried, never LLM-chosen.

In production against a streamable-http server, OAuth identity attaches to the TRANSPORT: pass an httpx.Auth as HttpServer(auth=...) and the adapter’s persistent httpx client handles token exchange, expiry, and 401-triggered refresh — without a reconnect. The battery ships the machine-to-machine default:

from neograph_mcp import HttpServer, client_credentials_auth

crm = HttpServer(
    url="https://mcp.example.com/mcp",
    auth=client_credentials_auth(
        server_url="https://mcp.example.com/mcp",
        client_id="my-service",
        client_secret=SECRET,        # from your secret manager, not source
        scopes="crm.read crm.write",
    ),
)

client_credentials_auth wraps the MCP SDK’s ClientCredentialsOAuthProvider (RFC 8414 discovery + token exchange + refresh — nothing re-implemented). For key-based client auth (private-key JWT / RFC 7523, on-behalf-of), wrap the SDK’s PrivateKeyJWTOAuthProvider / RFC7523OAuthClientProvider from mcp.client.auth.extensions.client_credentials the same way; for the interactive authorization-code flow, use the SDK’s OAuthClientProvider directly (it needs a redirect handler). When both auth and a bearer token_provider are set, auth wins — an Auth manages its own Authorization header. token_provider remains the right tool for stdio and static-bearer setups. Either way the token lives in the adapter’s httpx client — never in neograph state, the checkpoint, or the schema fingerprint. Do not assume header auth works over stdio; it does not.

Beat 4 — gated mutation

gate_tools_when on the action node pauses the run before update_deal fires. Because an act node’s tool step is a real checkpointed superstep, nothing has happened yet when the human is asked:

config = {"configurable": {"thread_id": "deal-1", "mcp_auth": "operator-A"}}

paused = await arun(action_graph, input={"node_id": "REVIEW-1"}, config=config)
# paused["__interrupt__"][0].value == {"pending_tool": "update_deal", "reason": "Approve CRM mutation?"}

# Approve → the tool runs exactly once:
result = await arun(action_graph, resume={"approved": True}, config=config)
# result["apply_action"].status == "applied"

Resume with a deny and the tool never runs — the agent is told why (a denial ToolMessage is fed back answering the pending call) and finalizes anyway:

result = await arun(action_graph, resume={"approved": False}, config=config)
# result["apply_action"].status == "blocked"  — update_deal never executed

This is the structural version of “approve before side effects”: you never touch the tool, and an MCP mutation cannot fire before a human approves. See the Human-in-the-Loop walkthrough for the full gate/interrupt story.

Beat 5 — typed results

An MCP tool that returns a schema-serializable value (a Pydantic model, or dict[str, Any]) sends its payload as structuredContent — plus a functionally-equivalent text block that the spec mandates purely for backwards compatibility. neograph is a structured-aware client: declare output_model= and it treats structuredContent as the single source of truth, rehydrating it into your model. ToolInteraction.typed_result then is that model:

class CrmSearchResult(BaseModel, frozen=True):
    hits: list[Deal]     # Deal(id, name, stage) — validated element-wise
    acting_as: str       # the echoed per-run identity

# declared once on the factory (see the top of the page):
#   output_models={"crm_search": CrmSearchResult, "kb_lookup": KbResult}

search = next(i for i in tool_log if i.tool_name == "crm_search")
deal = search.typed_result.hits[0]   # Deal(id='D1', name='Acme renewal', stage='negotiation')

No hand-parsing a JSON text block, and no shipping that block to the model either: because typed_result is a Pydantic model, the next ReAct turn’s ToolMessage is neograph’s BAML-style rendering of it (via describe_value), not a repr of raw content blocks. A tool with a bare -> dict/-> list server annotation emits no structuredContent; declaring output_model= on it raises a typed error naming the annotation fix. A server isError follows the adapter-native self-correcting path unchanged. Omit output_model= and the tool keeps its raw content-block result — the type channel is opt-in by declaration.

Read-only readers are marked idempotent=True, which flags them replay-safe: a read may be re-invoked to re-derive an expired resource, but a mutation may not (the default is the conservative non-idempotent).

The escape hatch: hand-rolled factories

The battery is an overridable default. A tool factory is just an async callable (config, tool_config) -> tool that owns its client — neograph core never holds an MCP session:

from langchain_mcp_adapters.client import MultiServerMCPClient

async def crm_search_factory(config, tool_config):
    token = config["configurable"].get("mcp_auth", "anon")
    client = MultiServerMCPClient({"crm": {
        "transport": "stdio", "command": sys.executable, "args": [DEMO_SERVER]}})
    tools = await client.get_tools(server_name="crm")
    tool = next(t for t in tools if t.name == "crm_search")
    original = tool.coroutine
    async def with_identity(**kw): return await original(**{**kw, "token": token})
    return tool.model_copy(update={"coroutine": with_identity})

compile(pipeline, tool_factories={"crm_search": crm_search_factory})

Same shape as DefaultPromptCompiler: a productized 90%-case battery over a seam you can always fill by hand.

One known tool, offline build: the singular factory

mcp_tool_factories(...) connects at build time — discovery is a connect. When you already know the tool’s name and the build path must stay offline (a compile() call in a test suite, a deployment step with no gateway access), declare it with the singular mcp_tool_factory(...) instead: it returns one factory with zero network I/O at construction — the connect is deferred into the factory body and fires on the agent’s first tool call.

Its second job is gateway federation. A gateway like IBM ContextForge re-exposes a peer’s tool namespaced as <peer>-<tool>, but your node binds a fixed bare name:

from neograph_mcp import mcp_tool_factory

factory = mcp_tool_factory(
    "crm", CRM,
    tool_name="crm-perplexity_research",  # the name the gateway exposes
    rename_to="perplexity_research",      # the bare name the node's Tool() binds
    token_provider=token_provider,        # identity: same path as the plural
)

@node(mode="agent", outputs=ResearchNote, model="research", prompt="research/note",
      tools=[Tool("perplexity_research", budget=1, idempotent=True)])
def research() -> ResearchNote: ...

graph = compile(pipeline, llm_factory=llm_factory, prompt_compiler=prompt_compiler,
                tool_factories={"perplexity_research": factory})
# Still no MCP connect. The first tool call inside arun() is where it happens —
# and a wrong tool_name fails loud there, listing what the server actually exposes.

The trade against the plural is deliberate: discovery fails fast at build; declaration fails loud at first call. Pick by what you know — the plural asks the server, the singular tells the server.

Bind vs compose: `mcp_session` for N calls over one connection

The factories bind one federated tool 1:1 — the agent decides when to call it. But some work is a composite: a scripted step that must issue several primitive calls itself and assemble the result deterministically (resolve a company by domain, then fetch its notes/emails/calls/meetings, then flatten). Teaching an agent to orchestrate that is non-deterministic and burns tool-calls; you want plain Python control flow. And binding-then-materializing each factory opens a fresh MCP session per call.

mcp_session is the seam for that: connect once, call N tools by name over the single connection, get typed/structured results, close. It reuses the same transport + token machinery as the factories, and — like mcp_tool_factory — is consumer-owned and offline-at-build (the connect defers to async with entry). Open it inside the node/tool body that uses it, in the same task; never store it.

from neograph_mcp import mcp_session

@node(mode="raw", outputs=DealReview)
async def deal_review(state, config):
    async with mcp_session("crm", CRM, token_provider=token_provider, config=config) as s:
        # Two primitives over ONE connection; identity minted once at entry.
        search = await s.call("crm_search", {"query": "Acme"}, output_model=CrmSearchResult)
        deal   = await s.call("get_deal", {"deal_id": search.hits[0].id})
    detail = json.loads(deal.text or "{}")          # McpCallResult: .text / .content / .structured
    return {"deal_review": DealReview(deal_id=search.hits[0].id, stage=detail["stage"], ...)}

call(name, args, output_model=Model) rehydrates the server’s structuredContent into your model (the same typed channel as the factory’s output_model=); without output_model it returns an McpCallResult with converted content blocks (.content), the raw structured dict (.structured), and a .text convenience. A server error raises McpToolCallError; a bare -> dict/-> list tool with output_model= raises the typed missing-structuredContent error. Construction does no I/O; timeout= (default 30s) bounds the connect and each call so a hung server surfaces a TimeoutError instead of stalling. Discover names with await s.tool_names().

Bind when the agent chooses the call; compose when your code makes the calls. A gateway re-exposes federated tools namespaced (<peer>-<tool>) — a composite calls them by that name directly (no rename; there is no Tool() binding to reconcile).

Key takeaways

mcp_tool_factories(...) returns a dict — slice it per node for least-privilege binding
mcp_tool_factory(...) binds one declared tool with zero network at build — use it for offline compile/test paths and the gateway-federated <peer>-<tool> → bare Tool(name) rename
mcp_session(...) composes: one connection, N calls by name, typed/structured results (McpCallResult / output_model=), a timeout guard — use it for a scripted composite that issues the primitive calls itself
Each node’s tools= (and tool_factories=) is its privilege boundary; a research agent with no update_deal cannot mutate
MCP tools are async-only — drive with arun(); lint() flags tool_requires_async_driver for a sync run()
token_provider reads config['configurable']['mcp_auth'] for per-run identity — a tool argument over stdio, a bearer header (httpx.Auth) over streamable-http
gate_tools_when pauses before a mutation fires; approve runs it exactly once, deny never runs it and the agent is told why
ToolInteraction.typed_result carries the structured MCP payload; mark read-only readers idempotent=True

See examples/23_mcp_client_selective_binding.py for the full runnable example (uv run --extra dev --extra mcp-examples python examples/23_mcp_client_selective_binding.py), examples/25_mcp_single_tool_gateway.py for the singular-factory story (offline build, gateway rename, per-run identity), and examples/26_mcp_composite_session.py for the composite story (mcp_session: one connection, two federated primitives, a typed assembled result) — all against the same demo server.