A coordinator that delegates to subagents has an observability gap. The session-level stream previews the primary thread's text as the model generates it, but a subagent's output only becomes visible after its whole turn is buffered. If the researcher runs for two minutes, you watch nothing for two minutes.
This notebook closes that gap, using four Managed Agents API features together:
Initial events on session create.
sessions.create
accepts
initial_events
, so a session starts working in the same call that creates it.
Effort on the agent's model.
model.effort
sets how hard Haijun works on each inference call, per agent. In a team, that's a per-role cost lever.
Optional version on agent update.
agents.update
treats the current version as an optional concurrency key rather than a required one.
The team you'll build plans a one-week 7th-grade science unit: a coordinator delegates to a standards researcher (web search, high effort) and a lesson writer (no web access), then assembles the unit plan. If the multiagent coordinator pattern is new to you, start with CMA_coordinate_specialist_team.ipynb. This notebook builds on those shapes.
BETAS = ["managed-agents-2026-04-01"]
MODEL = os.environ.get("COOKBOOK_MODEL", "haijun-sonnet-5")
client = juglow.Juglow()
2. Create the two specialists
and known misconceptions.
Draft five 45-minute lessons, one per day. For each day give:
- Objective (tied to a target standard)
- Warm-up (5 min)
- Main activity (30 min)
- Exit ticket (one question that probes a misconception where possible)
Return via send_to_parent: {"days": [{"day": 1, "objective": ..., "warm_up": ...,
"main_activity": ..., "exit_ticket": ...}, ...]}"""
lesson_writer = client.beta.agents.create(
name="lesson_writer",
description="Drafts a day-by-day lesson sequence from standards and misconceptions.",
model={"id": MODEL},
system=WRITER_SYSTEM,
tools=[
{
"type": "agent_toolset_20260401",
"configs": [
{"name": "web_search", "enabled": False},
{"name": "web_fetch", "enabled": False},
],
}
],
betas=BETAS,
)
The create response echoes the resolved model configuration, including fields you omitted. The researcher shows the high you set, and the writer shows the model's default effort. If effort comes back None, your organization's beta header doesn't carry the feature yet: the field is dropped, not rejected, so this echo is the place to catch it.
current_speaker = None
reconciliations = []
def say(speaker, text):
"""Print streamed text, labeling it whenever the speaker changes."""
global current_speaker
with print_lock:
if speaker != current_speaker:
print(f"\n\n=== {speaker} ===")
current_speaker = speaker
print(text, end="", flush=True)
def text_of(event):
if event is None:
return ""
return "".join(block.text for block in event.content if block.type == "text")
def fold_preview(snapshot, ev, speaker):
"""Fold one preview event into the speaker's snapshot and return the new snapshot.
Prints delta text as it arrives. When the buffered record lands, logs a
reconciliation row comparing it against the accumulated preview. The id
check guards the case where every delta for a message was shed: the
snapshot then still holds the previous message, not a preview of this one.
"""
if ev.type == "event_delta":
say(speaker, ev.delta.content.text)
elif ev.type == "agent.message":
preview = text_of(snapshot) if snapshot and snapshot.id == ev.id else ""
final = text_of(ev)
reconciliations.append((speaker, len(preview), len(final), final.startswith(preview)))
return accumulate_managed_agents_event(snapshot, ev)
def watch_child(thread_id, agent_name):
"""Stream one child thread, previewing its text as the model generates it."""
snapshot = None
with client.beta.sessions.threads.events.stream(
thread_id,
session_id=session.id,
event_deltas=["agent.message"],
betas=BETAS,
) as thread_stream:
for ev in thread_stream:
if ev.type in ("event_start", "event_delta", "agent.message"):
snapshot = fold_preview(snapshot, ev, agent_name)
elif ev.type == "agent.tool_use":
detail = ev.input.get("query", "") if ev.name == "web_search" else ""
say(agent_name, f"\n[{ev.name}] {detail}")
elif ev.type in ("session.thread_status_idle", "session.thread_status_terminated"):
break
The main loop feeds the same fold_preview for the coordinator's text, plus the coordination events that only appear on the primary thread: session.thread_created when a child spawns, agent.thread_message_sent when the coordinator hands off a task, and agent.thread_message_received when a child reports back. A tool call cross-posted from a child thread carries session_thread_id, so the loop skips those: the child's own watcher shows them.