Run an isolated child agent#
Run a child with read-only tools and explicitly shared temporary storage. Tool restrictions are not an operating-system sandbox.
Run it#
From the repository root:
uv run python docs/_examples/subagent.py
The output includes:
Two isolated sessions; child links to parent
This program uses deterministic offline models and temporary storage where needed. It requires no API key and is executed by the documentation test suite.
Choose the child’s capabilities explicitly#
Give a child profile its own instructions, model, and tools. The parent can delegate work to the profile without giving it every capability the parent has:
async def main() -> None:
with TemporaryDirectory(prefix="zett-subagent-") as directory:
persistence = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
try:
definition = SubAgentDefinition(
name="review",
description="review a design with read-only tools",
system_prompt="Review the supplied design and return a concise report.",
model=ChildModel(),
extensions=(persistence, FileSystemExtension(read_only=True), ToolGuidelinesExtension()),
)
client = await create_agent(
ParentModel(),
config=AgentRunConfig(session_id="parent"),
extensions=[persistence, SubAgentExtension([definition]), ToolGuidelinesExtension()],
)
print((await client.run("Review the design")).content)
sessions = await persistence.list_sessions()
children = [session for session in sessions if session.parent_session_id == "parent"]
assert len(sessions) == 2 and len(children) == 1
assert children[0].session_id != "parent"
print("Two isolated sessions; child links to parent")
finally:
await persistence.close()
The example stores the parent and child as separate sessions in the same temporary database. A parent-session link lets your application show how they relate. Read-only filesystem tools restrict the exposed operations; they do not isolate the process or prevent arbitrary code from writing.
Complete source#
1"""Delegate a task to a child profile and inspect its durable parent linkage."""
2
3import asyncio
4import json
5from collections.abc import AsyncIterator
6from pathlib import Path
7from tempfile import TemporaryDirectory
8
9from zett_agent.agent import AgentRunConfig
10from zett_agent.client import create_agent
11from zett_agent.extensions.file_system import FileSystemExtension
12from zett_agent.extensions.sqlite import SQLiteSessionExtension
13from zett_agent.extensions.subagent import (
14 SubAgentDefinition,
15 SubAgentExtension,
16)
17from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension
18from zett_agent.messages import (
19 AssistantMessage,
20 ToolCall,
21 ToolMessage,
22)
23from zett_agent.model import (
24 ModelEvent,
25 ModelRequest,
26 ModelResponse,
27 RetryOptions,
28)
29
30
31class ParentModel:
32 retry = RetryOptions(max_retries=0)
33
34 async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
35 last = request.messages[-1]
36 if isinstance(last, ToolMessage):
37 report = json.loads(last.text)
38 message = AssistantMessage(content=f"Child report: {report['content']}")
39 else:
40 message = AssistantMessage(
41 tool_calls=(
42 ToolCall(
43 "task-1",
44 "task",
45 {"description": "Review design", "prompt": "Review the design", "subagent_type": "review"},
46 ),
47 )
48 )
49 yield ModelEvent.completed(ModelResponse(message))
50
51
52class ChildModel:
53 retry = RetryOptions(max_retries=0)
54
55 async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
56 assert {tool.name for tool in request.tools} == {"read_file", "view_image", "glob", "grep"}
57 yield ModelEvent.completed(ModelResponse(AssistantMessage(content="Keep the runtime small.")))
58
59
60async def main() -> None:
61 with TemporaryDirectory(prefix="zett-subagent-") as directory:
62 persistence = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
63 try:
64 definition = SubAgentDefinition(
65 name="review",
66 description="review a design with read-only tools",
67 system_prompt="Review the supplied design and return a concise report.",
68 model=ChildModel(),
69 extensions=(persistence, FileSystemExtension(read_only=True), ToolGuidelinesExtension()),
70 )
71 client = await create_agent(
72 ParentModel(),
73 config=AgentRunConfig(session_id="parent"),
74 extensions=[persistence, SubAgentExtension([definition]), ToolGuidelinesExtension()],
75 )
76 print((await client.run("Review the design")).content)
77 sessions = await persistence.list_sessions()
78 children = [session for session in sessions if session.parent_session_id == "parent"]
79 assert len(sessions) == 2 and len(children) == 1
80 assert children[0].session_id != "parent"
81 print("Two isolated sessions; child links to parent")
82 finally:
83 await persistence.close()
84
85
86if __name__ == "__main__":
87 asyncio.run(main())
Next: Compose built-in extensions for the subagent profile options, or Iterate toward a goal for evaluation-driven continuation.