Recipe: Local MCP Tools¶

Goal¶

Mount local MCP tools using FastMCP objects, scripts, commands, or pre-built sessions.

Example (MCP script config)¶

from doc_processing_toolkit import Agent

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="Use calculator tools when needed.",
    tools=[
        {
            "type": "mcp_script",
            "path": "tests/fixtures/sample_mcp_server.py",
            "name": "calculator",
        }
    ],
)

result = await agent.run("Multiply 6 and 7")
print(result.output)

Command-based local server example:

agent = Agent(
    model="gpt-4o",
    instructions="Use command-spawned MCP tools.",
    tools=[
        {
            "type": "mcp_command",
            "command": "python",
            "args": ["-m", "tests.fixtures.sample_mcp_server"],
            "name": "command-server",
        }
    ],
)

Environment for a subprocess server¶

A server started by mcp_script or mcp_command runs in its own process, and the MCP SDK gives that process only a short allow-list of your environment (PATH, the home directory and a few others; the list differs by platform). Until 0.5.3 nothing else reached it: a std-tools skills server read no DPTTOOLS_SKILLS_DIR and fell back to ./.agent_skills, and a web-search server had no API key. Name every variable the server needs in env; it is merged over the allow-list, so PATH survives. cwd sets the directory the server starts in. The values are never logged.

import os, sys

agent = Agent(
    model="claude-sonnet-4-5",
    tools=[
        {
            "type": "mcp_command",
            "command": sys.executable,
            "args": ["-m", "doc_processing_std_tools"],
            "name": "std-tools",
            "env": {
                "DPTTOOLS_SKILLS_DIR": "/srv/skills",
                "DPTTOOLS_WEB_SEARCH_API_KEY": os.environ["DPTTOOLS_WEB_SEARCH_API_KEY"],
            },
            "cwd": "/srv",
        }
    ],
)

The same keys work in a YAML tools: entry (env: as a mapping, cwd: as a string). A non-mapping env, a None value (the usual slip is os.environ.get(...) on an unset variable, which would reach the server as the text None) or a non-path cwd fails at Agent(...), not at the first run. Relative args of an mcp_command resolve against cwd; an mcp_script path is resolved against this process's directory first, so it keeps working when cwd is set. agent.to_config() keeps the values out of the recipe — the entry is emitted without env, and the variable names are listed under unserialisable.tool_env so the host supplies them again.

Exposing only some of a server's tools¶

Give the config an allowed_tools list (0.5.1). The model is shown only those tools; a listed name the server lacks logs one WARNING:

tools=[
    {
        "type": "mcp_script",
        "path": "tests/fixtures/sample_mcp_server.py",
        "name": "calculator",
        "allowed_tools": ["multiply"],
    }
]

The same key works on mcp_command configs and pre-built sessions. A FastMCP object takes it through the dict form:

from mcp.server.fastmcp import FastMCP

server = FastMCP("calculator")
# ... @server.tool() definitions ...

tools=[{"type": "fastmcp", "server": server, "allowed_tools": ["multiply"]}]

See Choosing which tools a server exposes.

Reusing one server across runs¶

The server is started when the Agent first connects and is reused by every later run() and every AgentSession turn on that event loop — one subprocess, one tool listing:

async with Agent(
    model="claude-sonnet-4-5",
    instructions="Use calculator tools when needed.",
    tools=[{"type": "mcp_script", "path": "tests/fixtures/sample_mcp_server.py", "name": "calculator"}],
) as agent:
    first = await agent.run("Multiply 6 and 7")
    second = await agent.run("Now add 8 to that")  # same server process, same tools

close() stops the server but keeps its config, so an Agent used after close() reconnects rather than running toolless. run_sync() starts and stops the server inside its own event loop, one process per call — keep a long-lived loop and await agent.run(...) when that matters.

Pre-built session (async only)¶

A session you opened yourself belongs to the task and loop that created it, so it has to stay open around the whole run:

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client(REMOTE_URL, headers=HEADERS) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        agent = Agent(
            model="claude-sonnet-4-5",
            tools=[{"session": session, "name": "calculator"}],
        )
        result = await agent.run("Multiply 6 and 7")

run_sync() cannot use a pre-built session: it opens a fresh event loop per call, where a session created elsewhere is already dead. Passing the same session object twice loads its tools once. To have the toolkit own a remote connection instead — with run_sync support and automatic reconnects — use tool format 9, client-side remote MCP.

What to validate¶

tool names from local server appear in result.tools_used
output contains expected computed value
server lifecycle starts/stops cleanly during run
a second run on the same Agent reuses the server instead of starting another one
with allowed_tools, only the listed tools appear in the model's tool list (a run's tools_used never names another)
a variable named in env is visible to the server, and one that is not named is absent (tests/unit/test_mcp_subprocess_env.py drives a real subprocess through both)

Test sources¶

tests/e2e/test_tool_formats.py (test_format_3_fastmcp_server, test_format_4_mcp_script_path, test_format_5_mcp_command, test_format_7_prebuilt_session)
tests/integration/test_anthropic_function_calling.py (TestAnthropicFunctionCallingWithLocalMCP)
tests/unit/test_mcp_session_lifecycle.py (connect once per loop, symmetric teardown, reuse after close())