Agent Composition & Reusable Primitives
Simple agents start as a single file. One system prompt, one tool list, one orchestration loop, one model — all wired together directly. This works fine until the second agent arrives. Then the third. Suddenly you have five agents that each implement their own retry logic, their own tool registration, their own context assembly, and their own model client initialization. The code is not bad — it is duplicated, and duplication in agent systems compounds faster than in traditional software because each agent carries a heavier configuration surface.
The fix is the same one software engineering discovered decades ago: build from composable units with clean interfaces. But agents add a twist. A traditional function takes arguments and returns values. An agent takes natural language, makes nondeterministic decisions, calls external tools, accumulates state, and produces output whose shape depends on what happened during execution. Composing agents means composing unpredictable processes, and the interface contracts need to account for that.
The Monolith Trap #
Agent monoliths emerge naturally. You start with a prototype, and the fastest path to a working demo is to wire everything together inline.
# The monolith: everything hardcoded in one place
async def handle_request(user_message: str) -> str:
client = ModelClient(model="large-reasoning-v2", api_key=API_KEY)
tools = [
search_web,
read_database,
send_email,
generate_chart,
]
memory = ConversationMemory(max_tokens=8000)
memory.add("user", user_message)
system_prompt = """You are a research assistant.
Use the search tool to find information.
Use the database tool to look up internal records.
..."""
while True:
response = await client.chat(
system=system_prompt,
messages=memory.messages,
tools=tools,
)
memory.messages.append(response.as_message())
if response.has_tool_calls:
for call in response.tool_calls:
result = await execute_tool(call)
memory.messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": call.id,
})
else:
return response.text
This is readable and it works. But it has four problems that surface the moment you try to scale beyond one agent:
Hardcoded dependencies. The model, the tool set, and the memory implementation are all chosen inside the function. You cannot swap the model for a cheaper one in tests, inject a mock tool set, or change the memory strategy without editing the function body.
Opaque interface. The function takes a string and returns a string. The caller has no visibility into what tools were called, how many model turns it took, how many tokens were consumed, or whether the agent encountered and recovered from errors.
No reuse across agents. The orchestration loop — the while True cycle of calling the model, executing tools, and feeding results back — is a general pattern. But it is baked into this specific agent with this specific prompt and these specific tools.
No composability. You cannot use this agent as a tool inside another agent, because its interface does not match what a tool call expects. You cannot chain it with another agent without writing glue code that is specific to both.
The Agent as a Callable Unit #
The first step toward composition is defining a clean boundary around an agent. An agent needs an interface — something that describes what goes in, what comes out, and what side effects it might produce.
import asyncio
import json
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any, Protocol
@dataclass
class AgentInput:
"""What goes into an agent invocation."""
task: str
context: dict[str, Any] = field(default_factory=dict)
max_steps: int = 20
timeout_seconds: float = 120.0
max_tokens: int | None = None
@dataclass
class AgentOutput:
"""What comes out of an agent invocation."""
result: str
steps_taken: int = 0
tokens_used: int = 0
tool_calls: list[dict] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
success: bool = True
error: str | None = None
@dataclass
class ModelUsage:
input_tokens: int = 0
output_tokens: int = 0
@property
def total_tokens(self) -> int:
return self.input_tokens + self.output_tokens
@dataclass
class ToolCall:
id: str
name: str
arguments: dict[str, Any]
@dataclass
class ModelResponse:
text: str = ""
tool_calls: list[ToolCall] = field(default_factory=list)
usage: ModelUsage = field(default_factory=ModelUsage)
stop_reason: str = "end_turn"
@property
def has_tool_calls(self) -> bool:
return bool(self.tool_calls)
def as_message(self) -> dict[str, Any]:
"""Return the provider-neutral assistant history entry."""
message: dict[str, Any] = {
"role": "assistant",
"content": self.text,
}
if self.tool_calls:
message["tool_calls"] = [
{
"id": call.id,
"name": call.name,
"arguments": call.arguments,
}
for call in self.tool_calls
]
return message
class AgentCallable(Protocol):
"""The minimal interface every agent exposes."""
async def run(self, input: AgentInput) -> AgentOutput: ...
This protocol gives you three things. First, typed inputs and outputs — callers know exactly what they can pass in and what they will get back. Second, observability by default — every invocation produces metadata about steps, tokens, and tool usage. Third, composability — any function that accepts an AgentCallable can work with any agent that implements the protocol, regardless of what model, tools, or orchestration strategy it uses internally.
The AgentOutput structure is deliberately richer than a bare string. When you compose agents — chaining one's output into another's input, or using an agent as a tool — the caller often needs more than the text result. It needs to know whether the agent succeeded, how much budget it consumed, and what tools it invoked. Burying that information inside the agent forces callers to parse text or add out-of-band telemetry to recover it.
Dependency Injection #
The monolith hardcodes its dependencies. A composable agent receives them.
# A tool is just a callable with metadata
@dataclass
class ToolDefinition:
name: str
description: str
parameters: dict # JSON Schema
fn: Callable[..., Awaitable[Any]]
class Memory(Protocol):
messages: list[dict[str, Any]]
class ConversationMemory:
"""Minimal in-memory history used by the examples."""
def __init__(self, max_tokens: int | None = None):
self.messages: list[dict[str, Any]] = []
self.max_tokens = max_tokens
# A model client hides the provider behind a stable interface
class ModelClient(Protocol):
async def chat(
self,
system: str,
messages: list[dict],
tools: list[ToolDefinition] | None = None,
temperature: float = 0.0,
max_tokens: int = 4096,
) -> "ModelResponse": ...
class Agent:
"""An agent assembled from injected components."""
def __init__(
self,
model: ModelClient,
tools: list[ToolDefinition],
system_prompt: str,
memory_factory: Callable[[], Memory] | None = None,
):
self.model = model
self.tools = tools
self.system_prompt = system_prompt
self.memory_factory = memory_factory or (lambda: ConversationMemory())
async def run(self, input: AgentInput) -> AgentOutput:
try:
return await asyncio.wait_for(
self._run(input),
timeout=input.timeout_seconds,
)
except asyncio.TimeoutError:
return AgentOutput(
result="Agent timed out",
success=False,
error=f"Exceeded timeout of {input.timeout_seconds} seconds",
)
async def _run(self, input: AgentInput) -> AgentOutput:
memory = self.memory_factory()
user_message = input.task
if input.context:
context = json.dumps(input.context, default=str, indent=2)
user_message += f"\n\nContext:\n{context}"
memory.messages.append({"role": "user", "content": user_message})
steps = 0
total_tokens = 0
all_tool_calls = []
while steps < input.max_steps:
if input.max_tokens is not None and total_tokens >= input.max_tokens:
return AgentOutput(
result="Token budget exhausted",
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=False,
error="Exceeded token budget",
)
max_output_tokens = 4096
if input.max_tokens is not None:
max_output_tokens = min(
max_output_tokens,
max(1, input.max_tokens - total_tokens),
)
response = await self.model.chat(
system=self.system_prompt,
messages=memory.messages,
tools=self.tools,
max_tokens=max_output_tokens,
)
total_tokens += response.usage.total_tokens
steps += 1
memory.messages.append(response.as_message())
if input.max_tokens is not None and total_tokens > input.max_tokens:
return AgentOutput(
result=response.text or "Token budget exceeded",
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=False,
error="Exceeded token budget",
)
if response.has_tool_calls:
for call in response.tool_calls:
result = await self._execute_tool(call)
all_tool_calls.append({
"tool": call.name,
"args": call.arguments,
"result_preview": str(result)[:200],
})
memory.messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": call.id,
})
else:
succeeded = response.stop_reason in {"end_turn", "stop_sequence"}
return AgentOutput(
result=response.text,
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=succeeded,
error=None if succeeded else f"Model stopped: {response.stop_reason}",
)
return AgentOutput(
result="Max steps reached",
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=False,
error="Exceeded maximum step count",
)
async def _execute_tool(self, call: ToolCall) -> Any:
tool = next((t for t in self.tools if t.name == call.name), None)
if tool is None:
return f"Error: unknown tool '{call.name}'"
return await tool.fn(**call.arguments)
Now the same Agent class powers completely different agents just by varying what you inject:
# A research agent with web tools and a large model
research_agent = Agent(
model=large_model_client,
tools=[search_web_tool, fetch_page_tool, summarize_tool],
system_prompt="You are a research assistant. Find and synthesize information.",
)
# A code review agent with repository tools and a fast model
review_agent = Agent(
model=fast_model_client,
tools=[read_file_tool, list_files_tool, run_tests_tool],
system_prompt="You are a code reviewer. Analyze the code for bugs and style issues.",
)
# A test configuration that injects mocks
test_agent = Agent(
model=mock_model_client,
tools=[mock_search_tool],
system_prompt="You are a test agent.",
)
The orchestration loop is the same. What varies is the model, the tools, and the prompt. Dependency injection separates the stable machinery from the variable configuration.
Model Abstraction Boundaries #
Different providers have different APIs, different token counting, different tool call formats, and different streaming behaviors. Hiding all of that behind a single interface means your agents do not care which provider is behind the call.
from anthropic import AsyncAnthropic
class AnthropicModelClient:
"""Adapter for one specific provider's API."""
def __init__(self, api_key: str, model: str):
self.client = AsyncAnthropic(api_key=api_key)
self.model = model
async def chat(
self,
system,
messages,
tools=None,
temperature=0.0,
max_tokens=4096,
):
tool_defs = [self._convert_tool(t) for t in (tools or [])]
response = await self.client.messages.create(
model=self.model,
system=system,
messages=self._convert_messages(messages),
tools=tool_defs or None,
temperature=temperature,
max_tokens=max_tokens,
)
return self._normalize_response(response)
def _convert_tool(self, tool: ToolDefinition) -> dict:
return {
"name": tool.name,
"description": tool.description,
"input_schema": tool.parameters,
}
def _convert_messages(self, messages: list[dict]) -> list[dict]:
"""Convert provider-neutral history to Anthropic content blocks."""
converted = []
for message in messages:
role = message["role"]
if role == "assistant":
content = []
if message.get("content"):
content.append({"type": "text", "text": message["content"]})
content.extend(
{
"type": "tool_use",
"id": call["id"],
"name": call["name"],
"input": call["arguments"],
}
for call in message.get("tool_calls", [])
)
converted.append({"role": "assistant", "content": content})
elif role == "tool":
result = {
"type": "tool_result",
"tool_use_id": message["tool_call_id"],
"content": message["content"],
"is_error": message.get("is_error", False),
}
# Anthropic expects parallel tool results in one user message.
if (
converted
and converted[-1]["role"] == "user"
and isinstance(converted[-1]["content"], list)
and all(
block.get("type") == "tool_result"
for block in converted[-1]["content"]
)
):
converted[-1]["content"].append(result)
else:
converted.append({"role": "user", "content": [result]})
else:
converted.append({"role": "user", "content": message["content"]})
return converted
def _normalize_response(self, raw) -> ModelResponse:
text = []
tool_calls = []
for block in raw.content:
if block.type == "text":
text.append(block.text)
elif block.type == "tool_use":
tool_calls.append(ToolCall(
id=block.id,
name=block.name,
arguments=block.input,
))
return ModelResponse(
text="".join(text),
tool_calls=tool_calls,
usage=ModelUsage(
input_tokens=raw.usage.input_tokens,
output_tokens=raw.usage.output_tokens,
),
stop_reason=raw.stop_reason,
)
This adapter pattern is the same thing you would do for any external dependency. But in agent systems it matters more because you switch models frequently: a cheaper model for routing decisions, a larger one for complex reasoning, a specialized one for code generation. Each switch should be a configuration change.
Agent as a Tool #
Here is where composition gets interesting. If an agent conforms to a callable interface, you can wrap it as a tool and give it to another agent. The outer agent sees a tool; the inner agent sees a task. Neither knows or cares about the other's implementation.
def agent_as_tool(
agent: Agent,
name: str,
description: str,
parameter_description: str = "The task to perform",
) -> ToolDefinition:
"""
Wrap any agent behind a tool schema so it can be
invoked by another agent as a tool call.
"""
async def invoke(task: str) -> str:
output = await agent.run(AgentInput(task=task))
if not output.success:
return f"Agent failed: {output.error}"
return output.result
return ToolDefinition(
name=name,
description=description,
parameters={
"type": "object",
"properties": {
"task": {
"type": "string",
"description": parameter_description,
}
},
"required": ["task"],
},
fn=invoke,
)
Now an orchestrator agent can use specialized sub-agents the same way it uses any other tool:
# Create specialist agents
research = Agent(
model=large_model,
tools=[search_web_tool, fetch_page_tool],
system_prompt="You are a research specialist. Find accurate information.",
)
writer = Agent(
model=large_model,
tools=[],
system_prompt="You are a technical writer. Produce clear, concise prose.",
)
# Wrap them as tools
research_tool = agent_as_tool(
research,
name="research",
description="Research a topic and return a summary of findings",
parameter_description="The research question to investigate",
)
writing_tool = agent_as_tool(
writer,
name="write_draft",
description="Write a draft document given a topic and source material",
parameter_description="Instructions for what to write, including source material",
)
# The orchestrator sees tools, not agents
orchestrator = Agent(
model=large_model,
tools=[research_tool, writing_tool],
system_prompt="""You are a project coordinator.
Break tasks into research and writing steps.
Use the research tool to gather information.
Use the write_draft tool to produce the final output.""",
)
┌──────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ │
│ System prompt: "You are a project coordinator..." │
│ Model: large-reasoning-v2 │
│ │
│ Tools: │
│ ┌────────────────────┐ ┌────────────────────────┐ │
│ │ research (tool) │ │ write_draft (tool) │ │
│ │ ┌──────────────┐ │ │ ┌──────────────────┐ │ │
│ │ │ Research │ │ │ │ Writer Agent │ │ │
│ │ │ Agent │ │ │ │ │ │ │
│ │ │ model: large │ │ │ │ model: large │ │ │
│ │ │ tools: │ │ │ │ tools: (none) │ │ │
│ │ │ - search │ │ │ │ │ │ │
│ │ │ - fetch │ │ │ │ │ │ │
│ │ └──────────────┘ │ │ └──────────────────┘ │ │
│ └────────────────────┘ └────────────────────────┘ │
└──────────────────────────────────────────────────────┘
This pattern — agent-as-tool — is the fundamental composition primitive. It works because the tool interface is the universal contract that models already understand. The model does not need to know that research is backed by another agent rather than a simple function. It just sees a tool with a name, a description, and parameters.
Controlling the Inner Agent #
Wrapping an agent as a tool raises practical concerns. The inner agent might run for a long time, consume excessive tokens, or enter an infinite loop. The outer agent has no visibility into this — it just sees a tool call that takes a while to return.
The solution is to enforce budgets at the boundary:
def agent_as_tool(
agent: Agent,
name: str,
description: str,
max_steps: int = 10,
timeout_seconds: float = 60.0,
max_tokens: int = 50_000,
) -> ToolDefinition:
"""
Wrap an agent as a tool with explicit resource boundaries.
"""
async def invoke(task: str) -> str:
output = await agent.run(AgentInput(
task=task,
max_steps=max_steps,
timeout_seconds=timeout_seconds,
max_tokens=max_tokens,
))
if output.tokens_used > max_tokens:
return f"Agent exceeded token budget ({output.tokens_used}/{max_tokens})"
if not output.success:
return f"Agent failed after {output.steps_taken} steps: {output.error}"
return output.result
return ToolDefinition(
name=name,
description=description,
parameters={
"type": "object",
"properties": {
"task": {"type": "string", "description": "The task to perform"}
},
"required": ["task"],
},
fn=invoke,
)
Every agent-as-tool boundary should have a step limit, a timeout, and a token budget. The step and timeout limits are hard boundaries. The token limit stops additional model calls and caps the next call's output allowance; because providers report input usage after a request, the final call can still take the measured total slightly past the threshold.
Reusable Primitives #
Below the agent level, there are smaller building blocks that recur across nearly every agent you build. Extracting these into reusable primitives eliminates duplication and makes each agent's configuration more declarative.
The Orchestration Loop #
The core loop — call model, execute tools, feed results back, repeat — is the same across almost all agents. Factor it out:
async def run_agent_loop(
model: ModelClient,
system_prompt: str,
tools: list[ToolDefinition],
messages: list[dict],
max_steps: int = 20,
) -> AgentOutput:
"""
The generic agent loop. Reusable across any agent
regardless of domain, tools, or model.
"""
steps = 0
total_tokens = 0
all_tool_calls = []
while steps < max_steps:
response = await model.chat(
system=system_prompt,
messages=messages,
tools=tools,
)
total_tokens += response.usage.total_tokens
steps += 1
messages.append(response.as_message())
if not response.has_tool_calls:
succeeded = response.stop_reason in {"end_turn", "stop_sequence"}
return AgentOutput(
result=response.text,
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=succeeded,
error=None if succeeded else f"Model stopped: {response.stop_reason}",
)
for call in response.tool_calls:
tool = next((t for t in tools if t.name == call.name), None)
if tool is None:
messages.append({
"role": "tool",
"content": f"Unknown tool: {call.name}",
"tool_call_id": call.id,
})
continue
result = await tool.fn(**call.arguments)
all_tool_calls.append({
"tool": call.name,
"args": call.arguments,
})
messages.append({
"role": "tool",
"content": str(result),
"tool_call_id": call.id,
})
return AgentOutput(
result="Max steps reached",
steps_taken=steps,
tokens_used=total_tokens,
tool_calls=all_tool_calls,
success=False,
error="Exceeded maximum step count",
)
Now building an agent is similar to assembling configuration:
async def research_agent(task: str) -> AgentOutput:
return await run_agent_loop(
model=large_model_client,
system_prompt=RESEARCH_SYSTEM_PROMPT,
tools=[search_web_tool, fetch_page_tool],
messages=[{"role": "user", "content": task}],
max_steps=15,
)
Tool Factories #
Tools share patterns too. A database tool, a file-reading tool, and an API tool all need error handling, input validation, and result truncation. Factor the common wrapper:
import inspect
def make_tool(
name: str,
description: str,
parameters: dict,
fn: Callable[..., Any],
max_result_length: int = 4000,
) -> ToolDefinition:
"""
Wrap a raw function into a tool with standardized
error handling and output truncation.
"""
async def safe_fn(**kwargs) -> str:
try:
if inspect.iscoroutinefunction(fn):
result = await fn(**kwargs)
else:
# Keep blocking synchronous tools off the event loop.
result = await asyncio.to_thread(fn, **kwargs)
if inspect.isawaitable(result):
result = await result
text = str(result)
if len(text) > max_result_length:
text = text[:max_result_length] + f"\n... (truncated, {len(text)} chars total)"
return text
except Exception as e:
return f"Tool error: {type(e).__name__}: {e}"
return ToolDefinition(
name=name,
description=description,
parameters=parameters,
fn=safe_fn,
)
Every tool you build through make_tool automatically gets error isolation (a failing tool does not crash the agent loop) and output truncation (a tool that returns a 500KB response does not blow out the context window). These are not features you want to re-implement per tool.
Prompt Templates #
System prompts follow patterns too. A common structure is: role description, constraints, available context, output format instructions. Making this composable:
@dataclass
class PromptSection:
heading: str
content: str
condition: bool = True # Include only when True
def assemble_prompt(sections: list[PromptSection]) -> str:
"""
Build a system prompt from conditional sections.
Sections with condition=False are excluded.
"""
parts = []
for section in sections:
if section.condition:
parts.append(f"## {section.heading}\n\n{section.content}")
return "\n\n".join(parts)
# Usage: different agents share common sections
common_sections = [
PromptSection("Output Format", "Respond in plain text. Be concise."),
PromptSection("Constraints", "Do not make up facts. Cite sources when possible."),
]
research_prompt = assemble_prompt([
PromptSection("Role", "You are a research assistant."),
*common_sections,
PromptSection("Tools", "Use the search tool to find information on the web."),
])
analysis_prompt = assemble_prompt([
PromptSection("Role", "You are a data analyst."),
*common_sections,
PromptSection("Tools", "Use the query tool to run SQL against the database."),
])
This is more useful than it looks. When you have twenty agents sharing the same output format rules and safety constraints, changing those rules means changing them in one place rather than hunting through twenty system prompt strings.
Interface Contracts for Agent Composition #
When agents compose — one calling another as a tool, or a pipeline chaining several in sequence — the interface contract between them matters more than it does for normal function calls. The contract needs to handle three things that regular function signatures do not:
Partial success. An inner agent might accomplish 80% of its task before hitting a limit. The outer agent needs to know whether the result is complete or partial so it can decide whether to retry, supplement, or proceed with what it has.
Resource accounting. Token consumption and step counts from inner agents should propagate upward so the outer agent (or the system) can enforce global budgets. Without this, nested agent calls create invisible cost sinks.
Error context. When a nested agent fails, the error message needs to be useful to the model that will process it. "Internal server error" tells the outer agent nothing. "Research agent failed: search tool returned 429 rate limit after 3 retries" tells it to try a different approach.
@dataclass
class CompositionResult:
"""Rich result type for agent-to-agent communication."""
result: str
is_complete: bool = True
tokens_consumed: int = 0
steps_consumed: int = 0
inner_tool_calls: list[dict] = field(default_factory=list)
warnings: list[str] = field(default_factory=list)
def as_tool_result(self) -> str:
"""Format for consumption by an outer agent's model."""
parts = [self.result]
if not self.is_complete:
parts.append("[Note: this result may be incomplete]")
if self.warnings:
parts.append(f"[Warnings: {'; '.join(self.warnings)}]")
return "\n".join(parts)
The outer agent's model sees a text result (because tool results are text), but that text carries signals about completeness and quality that the model can reason about. If the research tool returns a result tagged with "[Note: this result may be incomplete]", the model can decide to call it again with a narrower query or proceed with the partial information.
Testing Composable Agents #
Composable agents are significantly easier to test than monoliths, because you can swap every external dependency.
import pytest
class MockModelClient:
"""A model client that returns scripted responses."""
def __init__(self, responses: list["ModelResponse"]):
self.responses = iter(responses)
self.calls: list[list[dict]] = []
async def chat(
self,
system,
messages,
tools=None,
temperature=0.0,
max_tokens=4096,
):
self.calls.append([message.copy() for message in messages])
return next(self.responses)
@pytest.mark.asyncio
async def test_research_agent_handles_empty_results():
async def empty_search(query: str) -> str:
return "No results"
mock_model = MockModelClient([
# First call: model decides to search
ModelResponse(tool_calls=[
ToolCall(
id="call-search-1",
name="search",
arguments={"query": "quantum computing"},
)
]),
# Second call: model summarizes empty results
ModelResponse(text="No results found for quantum computing."),
])
mock_search = make_tool(
name="search",
description="Search the web",
parameters={
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
fn=empty_search,
)
agent = Agent(
model=mock_model,
tools=[mock_search],
system_prompt="You are a research assistant.",
)
output = await agent.run(AgentInput(task="Research quantum computing"))
assert output.success
assert output.steps_taken == 2
assert len(output.tool_calls) == 1
assert output.tool_calls[0]["tool"] == "search"
assert [message["role"] for message in mock_model.calls[1]] == [
"user",
"assistant",
"tool",
]
assert mock_model.calls[1][-1]["tool_call_id"] == "call-search-1"
With dependency injection, you test agent behavior — the decisions it makes given specific model responses — without calling real models or real tools. The mock model client replays a scripted sequence of responses, and the test verifies that the agent handles each one correctly.
This also lets you test composition itself: given an orchestrator agent with a mocked inner agent (wrapped as a tool), does the orchestrator correctly call the inner agent, handle its response, and produce a coherent final output?
@pytest.mark.asyncio
async def test_orchestrator_delegates_to_research():
async def research_answer(task: str) -> str:
return "The capital of France is Paris."
# The inner agent is a mock that always succeeds
mock_inner = make_tool(
name="research",
description="Research a topic",
parameters={
"type": "object",
"properties": {"task": {"type": "string"}},
"required": ["task"],
},
fn=research_answer,
)
mock_model = MockModelClient([
# Orchestrator calls the research tool
ModelResponse(tool_calls=[
ToolCall(
id="call-research-1",
name="research",
arguments={"task": "What is the capital of France?"},
)
]),
# Orchestrator formats the final answer
ModelResponse(text="Based on my research, the capital of France is Paris."),
])
orchestrator = Agent(
model=mock_model,
tools=[mock_inner],
system_prompt="You are a coordinator.",
)
output = await orchestrator.run(AgentInput(task="Find out the capital of France"))
assert output.success
assert "Paris" in output.result
Composition Cost #
Not every agent system benefits from composition. Composition adds indirection, and indirection has costs:
Debugging across boundaries. When something goes wrong in a nested agent call, you need to trace through multiple layers to find the root cause. A monolithic agent's trace is flat; a composed agent's trace is a tree.
Latency overhead. Each agent-as-tool invocation adds the overhead of a full agent loop — context assembly, model call, potentially multiple tool calls. If the inner task is simple, the overhead of spinning up a full agent for it can dominate the actual work.
Over-abstraction. Extracting a reusable primitive that is only used once adds code without adding value. A tool factory is worth it when you have twenty tools; it is overhead when you have three.
The guideline: compose when you have multiple agents sharing the same primitives, or when you need agents calling other agents. For a single agent with a handful of tools that will not be reused elsewhere, a straightforward implementation is fine. The composability patterns are there for when the system grows — and in agent systems, it grows faster than you expect.
Conclusion #
Agent composition is about drawing boundaries and defining contracts. The core ideas:
- Agents are callable units. Define a protocol with typed inputs and outputs so that every agent presents the same interface regardless of its internals.
- Inject dependencies, do not hardcode them. Models, tools, and memory should be passed in at construction time. This makes agents testable, swappable, and configurable per environment.
- Agent-as-tool is the composition primitive. Wrapping an agent behind a tool schema lets any other agent invoke it without knowing or caring that it is backed by another model loop. Enforce budgets at the boundary.
- Extract reusable building blocks. The orchestration loop, tool error handling, prompt assembly — these patterns recur across every agent. Factor them out once, use them everywhere.
- Rich interfaces beat bare strings. Return step counts, token usage, tool call logs, and success/failure signals alongside the text result. Callers — whether human systems or outer agents — need this information to make good decisions.
The payoff is the ability to build agent systems where adding a new capability means assembling existing pieces with a new prompt, rather than writing another monolith from scratch.