Recipe: Agent Factory MCP Server (Meta-Agent Pattern)¶

Goal¶

Use the Agent Factory MCP server as a tool so a main agent can create and run specialized child agents.

Example: meta-agent spawning a child agent¶

from doc_processing_toolkit import Agent
from doc_processing_toolkit.mcp_servers import agent_factory_server

meta_agent = Agent(
    model="gpt-4o",
    instructions=(
        "You orchestrate subtasks. "
        "Use create_agent when delegation is useful and report the child result."
    ),
    tools=[agent_factory_server],
    max_iterations=10,
)

result = await meta_agent.run(
    "Create a child agent with claude-haiku-4-5 and solve: What is 5 + 7?"
)

print(result.output)
print(result.tools_used)

Direct tool usage example¶

from doc_processing_toolkit.mcp_servers.agent_factory import create_agent

child_result = create_agent(
    model="claude-sonnet-4-5",
    instructions="You are a concise analyst.",
    task="Summarize the advantages of async I/O in one paragraph.",
    enable_web_search=False,
    detailed_costs=True,
)

print(child_result["output"])
print(child_result["cost"])  # None when the run raised after it started: spend unknown, not zero

When another Agent calls create_agent as a tool, the child run's cost is added to the calling agent's cost through the result's _dpt_spend marker (see Tool-reported spend). A child run that failed carries no marker, because its spend is unknown.

Persistent sub-agent sessions (A2A)¶

create_agent discards its agent after one task. When the sub-agent needs to remember earlier turns, create a session instead and address it by id:

from doc_processing_toolkit.mcp_servers.agent_factory import (
    create_agent_session,
    send_to_agent_session,
    get_agent_session_status,
)

created = await create_agent_session(
    model="claude-sonnet-4-5",
    instructions="You are a research assistant. Keep track of what you have already found.",
    initial_message="Research the 2026 EU battery regulations.",
    compaction=True,          # recommended for long-lived sessions
)
session_id = created["session_id"]

# Later turns remember everything above.
follow_up = await send_to_agent_session(session_id, "Now compare that to the US rules.")
print(follow_up["reply"])

status = get_agent_session_status(session_id)
print(status["run_count"], status["total_cost"])

Running a turn in the background¶

Pass wait=False to schedule a turn and return immediately, then poll:

import asyncio

await send_to_agent_session(session_id, "Draft the full report.", wait=False)

while get_agent_session_status(session_id)["busy"]:
    await asyncio.sleep(2)

print(get_agent_session_status(session_id)["last_reply_preview"])

A second send while a turn is still running returns status="busy" rather than racing two turns against the same history. To have it wait its turn instead, queue it:

await send_to_agent_session(session_id, "Draft the full report.", wait=False)
# Runs after the report turn finishes; returns this turn's own reply.
summary = await send_to_agent_session(session_id, "Now summarise it in three lines.", on_busy="queue")

Routing between two sessions¶

The toolkit does not connect sessions to each other — your code is the message bus, which is what lets you decide who may talk to whom:

researcher = (await create_agent_session(
    model="claude-sonnet-4-5", instructions="You research topics thoroughly."
))["session_id"]

critic = (await create_agent_session(
    model="gpt-5.6-terra", instructions="You critique research for gaps and bias."
))["session_id"]

findings = await send_to_agent_session(researcher, "Summarize solid-state battery progress.")
critique = await send_to_agent_session(critic, f"Critique these findings:\n{findings['reply']}")
revised = await send_to_agent_session(researcher, f"Address this critique:\n{critique['reply']}")

print(revised["reply"])

See A2A: Agent-to-Agent Sessions for the full tool reference, and Hooks and Steering for routing messages into a session that is already mid-run.

Sessions live until the process exits

The registry is in-process and has no close or evict tool, so each session holds its agent and client state for the lifetime of the host process. Create them deliberately.

Checkpointing a session across a server restart¶

from doc_processing_toolkit.mcp_servers.agent_factory import export_agent_session, import_agent_session

saved = await export_agent_session(session_id, path="checkpoints/reviewer.json")   # after every completed turn
# ... the server restarts ...
restored = await import_agent_session(
    model="claude-haiku-4-5", instructions="You review contracts.", path="checkpoints/reviewer.json"
)
restored["session_id"], restored["run_count"], restored["converted_from"]   # the same id, the turns so far, None

export_agent_session returns the envelope text when no path is given ("busy" while a turn runs) and takes the media / media_store policy of to_json; import_agent_session rebuilds the agent from the options create_agent_session takes (the rest through config_yaml), registers the session under the envelope's id or session_id=, and under another provider converts the history and reports it (converted_from, conversion_report_count; get_agent_session_status shows converted_from too).

Discovering models¶

Before a meta-agent picks a model for a child, it can ask the registry what exists. list_available_models and get_model_info read model_information.json only — no API call, no key needed — and are callable directly as well as through MCP:

from doc_processing_toolkit.mcp_servers.agent_factory import get_model_info, list_available_models

stt = list_available_models(modality="stt")
stt["openai"]["models"]         # ['gpt-transcribe', 'gpt-4o-transcribe-diarize', 'whisper-1']
stt["google"]["models"]         # ['gemini-3.5-transcribe']
stt["openai"]["model_details"]["whisper-1"]
# {'modality': 'stt', 'supports_diarization': False, 'supports_word_timestamps': True,
#  'supports_streaming_transcription': False, 'supports_keywords': False,
#  'supports_language_hints': False, 'session_kinds': ['transcription', 'translation']}

translators = list_available_models(session_kind="translation")
translators["openai"]["models"] # ['gpt-realtime-translate', 'whisper-1']

The result is grouped by provider. models is the flat name list; model_details holds a small discovery block per model. modality takes "text", "image", "tts", "stt" or "realtime" (a row with no modality counts as text); session_kind takes one of utils.SESSION_KINDS and matches any model that serves that kind among others. An unknown filter value returns empty lists, not everything.

get_model_info returns the full record for one model:

info = get_model_info("gpt-realtime")

info["deprecation"]
# {'date': '2027-01-20', 'replacement': 'gpt-realtime-2.1', 'note': "OpenAI's deprecations page ..."}
info["session_kinds"]                              # ['realtime', 'transcription']
info["supported_input_types"]                      # ['text', 'audio']
info["realtime"]["toolkit_features"]["translation_sessions"]   # True

Fields worth reading before choosing a model:

deprecation — {"date", "replacement", "note"} when the vendor has retired or superseded the row; date is None where a replacement is recommended but no shutdown is announced. Route new work to replacement.
realtime.toolkit_features — on a realtime row, what the toolkit's adapter supports for that provider: tool_dispatch, visual_input, transparent_reconnect, and whether dedicated transcription_sessions, translation_sessions and live_sessions can be opened. The full realtime block is left out of list_available_models to keep the listing small.
supported_input_types — what you may attach: text, audio, base64-image/png, video, ... A part outside this list is dropped (or refused with on_unsupported_part="raise").
session_kinds — the session families the model serves. A conversational realtime row lists transcription because its own socket transcribes; that does not make it a dedicated transcription session, so route on the kind the row declares.

An absent key is not false

Discovery omits a field the registry row does not carry rather than filling in None or False. Read every field with .get(), and treat "missing" as "the registry says nothing", not as "unsupported":

get_model_info("whisper-1")["capabilities"].get("supports_streaming_transcription")  # False — stated
get_model_info("gpt-5.6-luna")["capabilities"].get("supports_streaming_transcription")  # None — absent

whisper-1 states False on purpose, because the API accepts the flag and ignores it; a text row simply never mentions transcription. session_kinds is the one field that is always present. A model the registry does not know raises ValueError here (an error result over MCP).

See Discovery reports what kind of model a row is for the complete field list.

What to validate¶

create_agent appears in tools_used for the meta-agent run
child result payload includes output + token/cost fields + error status
delegation succeeds without manual MCP wiring beyond tools=[agent_factory_server]
a session's second turn reflects context from its first (history is retained)
send_to_agent_session on an unknown id returns status="not_found" rather than raising
a send during an in-flight wait=False turn returns status="busy", or runs next with on_busy="queue"
suggest_agent_config returns a recommendation on every provider, including when AGENT_FACTORY_SUGGESTION_MODEL names a Gemini model
when the suggestion sub-agent's own run fails, the returned error names that upstream failure rather than a JSON-parse message — a failed run used to surface as Expecting value: line 1 column 1, which said nothing about the real cause
list_available_models(modality="stt") lists only speech-to-text rows, and a field a row does not carry is absent from model_details, never None

Test sources¶

tests/e2e/test_agent_factory.py (test_meta_agent_spawns_child)
tests/e2e/test_agent_factory.py (test_create_agent_basic, test_suggest_agent_config)
tests/unit/test_a2a_sessions.py (TestCreateAgentSession, TestSendToAgentSession, TestNonBlockingRuns, TestGetAgentSessionStatus)
tests/unit/test_discovery_fields.py (discovery filters and fields)