AG2 Network — Workflow Adapter
is the orchestrated multi-party adapter. A declarative
describes who speaks first, what conditions fire on each turn, and when the channel terminates. It's the modern replacement for the classic
GroupChat + Agent.handoffs
pattern — turn-taking lives in the hub, not in an in-process speaker selector.
Prerequisite: read
first. This skill assumes you know
,
, the channel lifecycle, and the basic
/
flow.
When to use
Load this skill when the user asks for any of:
- "Two/three/N agents with conditional handoffs"
- "Pipeline / sequence of agents (researcher → writer → editor)"
- "Triage agent routes to specialists"
- "Drafter / reviewer feedback loop"
- "Coordinator + specialists / hierarchical orchestration"
- "Migrate from /
ConversableAgent.handoffs
- "I want context-driven routing (, )"
- Strict 1Q1R between two agents → (see ).
- Free-form 2-party chat with no turn order → (see ).
- Fixed round-robin with no conditions → (see ).
If the user is unsure, the rule of thumb:
need any condition more complex than "next speaker in a fixed list"? Use .Shape
| |
|---|
| Participants | 2+ |
| Turn order | Whatever says |
| Auto-close | Yes — when graph emits or is hit |
| Default view | NamedWindowedSummary(recent_n=N*2)
|
| Default expectations | , turn_within(600s, auto_close)
|
| Required knob | {"graph": <TransitionGraph.to_dict()>}
|
The expectation is strict: a stuck speaker auto-closes the channel after 10 minutes. Tune via
.
Workflow agents and the
attaches
by default. The plugin adds only the
identity-level cross-cutting tools —
/
/
/
/
. Channel-specific tools (notably
, plus your workflow handoff
s) come from
adapter.tools_for(client, metadata, state, participant_id)
, resolved per turn by the default handler and merged into
— and
the adapter offers none:
WorkflowAdapter.tools_for(...)
returns
. So a workflow agent never sees
; routing is entirely your
-decorated handoff functions (returning
, mutating context vars, or calling
).
Consequence: the old "mid-turn
races the round-end
" hazard
can't happen on a workflow channel — you do
not need
for correctness. Choose by capability instead:
- Keep the plugin (default) if the agent should also to a peer, look up / / , or read/write channel from outside the graph. ( / spin up separate channels — they never emit on the workflow channel, so they don't disturb its turn machinery.)
- for a genuinely bare agent — only the domain s you wrote. Tidy for pure pipeline workers and most tests.
python
# Bare pipeline worker — no cross-cutting network tools:
worker = await wc.register(agent, Passport(name="worker"), Resume(), attach_plugin=False)
The default handler wraps the LLM's plain-text reply into the workflow
either way. The examples below pass
because they're pure pipeline workers; drop it if your agents need
/
/ etc. Full detail:
ag2-network-tools-and-views
→ "Adapter-owned tools (
)".
Smallest example — pipeline via
python
import asyncio
from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import (
EV_CHANNEL_CLOSED,
Hub,
HubClient,
LocalLink,
Passport,
Resume,
TransitionGraph,
)
async def main() -> None:
config = AnthropicConfig(model="claude-sonnet-4-6")
hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)
link = LocalLink(hub)
rhc, whc, ehc = (HubClient(link, hub=hub) for _ in range(3))
researcher = await rhc.register(
Agent("researcher", prompt="Bullet-point three facts on the topic.", config=config),
Passport(name="researcher"), Resume(), attach_plugin=False,
)
writer = await whc.register(
Agent("writer", prompt="Turn the bullets into one paragraph.", config=config),
Passport(name="writer"), Resume(), attach_plugin=False,
)
editor = await ehc.register(
Agent("editor", prompt="Tighten the paragraph. Reply with the final text.", config=config),
Passport(name="editor"), Resume(), attach_plugin=False,
)
graph = TransitionGraph.sequence([
researcher.agent_id, writer.agent_id, editor.agent_id,
])
channel = await researcher.open(
type="workflow",
target=[writer.agent_id, editor.agent_id],
knobs={"graph": graph.to_dict()},
)
await channel.send("Topic: how does HTTPS work?")
close_env = await researcher.wait_for_channel_event(
channel_id=channel.channel_id,
predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED,
timeout=180.0,
)
print(f"closed: {close_env.event_data.get('reason')!r}") # 'sequence_complete'
for hc in (rhc, whc, ehc):
await hc.close()
await hub.close()
asyncio.run(main())
TransitionGraph.sequence([a, b, c])
builds three
FromSpeaker(x) → AgentTarget(next)
transitions plus a
TerminateTarget("sequence_complete")
default. After
speaks no transition matches, the default fires, and the channel auto-closes.
anatomy
python
@dataclass(slots=True)
class TransitionGraph:
initial_speaker: str # agent_id of the first speaker (must be a participant)
transitions: list[Transition] # ordered list, walked in priority order on every fold
default_target: TransitionTarget # consulted if no transition matches
max_turns: int | None = None # hard turn cap; force-terminates on reaching
@dataclass(slots=True)
class Transition:
when: TransitionCondition # evaluated against the just-accepted envelope + state
then: TransitionTarget # if when() is True, this resolves the next speaker
priority: int = 0 # LOWER runs first; ties break by insertion order
On every accepted substantive envelope (text or packet), the adapter walks the
list. The
first matching condition's target resolves the next speaker. If none match,
is consulted. A
resolves with
, which makes the adapter return
AdapterResult(next_state=CLOSING, ...)
and the hub posts
with that target's
.
is a separate, harder cap: once
reaches it the adapter closes the channel immediately with reason
— it does
not fall through to
(so a
default_target=TerminateTarget("…")
reason only appears when an actual turn produced no matching transition).
sorts ascending — a lower number is checked first. The adapter does
sorted(transitions, key=lambda t: t.priority)
then walks the result, so
(the default) beats
. If you want a rule consulted before the others, either put it at the top of the list (the sort is stable, so equal priorities keep list order) or give it a
negative priority. A common mistake: writing
on a
ContextEquals(...) → TerminateTarget(...)
rule expecting it to win — it gets pushed to the
end and a
fallback fires first instead, so the channel never terminates.
Gotcha — the kickoff is the 's first turn
When you do
agent.open(type="workflow", ...)
then
, that first send is folded as
's turn — it's an envelope from that agent, and
starts equal to
, so
accepts it and the graph immediately routes onward.
The 's never runs for that turn — the text you sent
is its contribution.
So if your design wants the
first agent that should actually think (the drafter, the researcher, the planner) to respond to the kickoff prompt, that agent must
not be
and must
not be the one calling
. The clean answer is a
as the seeder — a non-LLM participant (no
, no plugin) whose only job is to open the channel and post the brief. Make it
and route
FromSpeaker(user) → AgentTarget(drafter)
:
python
from ag2.network import HumanClient, Passport # plus the usual imports
user_hc = HubClient(link, hub=hub)
user = await user_hc.register_human(Passport(name="user", kind="human")) # ← register_human, not register
drafter = await drf_hc.register(drafter_agent, Passport(name="drafter"), Resume(), attach_plugin=False)
reviewer = await rev_hc.register(reviewer_agent, Passport(name="reviewer"), Resume(), attach_plugin=False)
graph = TransitionGraph(
initial_speaker=user.agent_id,
transitions=[
Transition(when=ContextEquals("approved", value=True), then=TerminateTarget("approved")),
Transition(when=FromSpeaker(user.agent_id), then=AgentTarget(drafter.agent_id)), # brief → drafter
Transition(when=FromSpeaker(drafter.agent_id), then=AgentTarget(reviewer.agent_id)),
Transition(when=FromSpeaker(reviewer.agent_id), then=AgentTarget(drafter.agent_id)), # revise loop
],
default_target=TerminateTarget("max_revisions"),
max_turns=12,
)
channel = await user.open(type="workflow", target=[drafter.agent_id, reviewer.agent_id], knobs={"graph": graph.to_dict()})
await channel.send("Brief: …") # consumed as `user`'s turn → routes to `drafter`, who drafts
# Later, when the channel closes: await user.next_envelope(predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED, timeout=300)
register_human(passport, *, resume=None, rule=None, auto_ack_invites=True)
returns a
(
now
rejects and points you here). It auto-acks invites so the quorum handshake completes without UI round-trips; pass
to gate joins by hand. Drive it with
(push) or
/
(pull) — see
→ "
".
If you'd rather not introduce a
, the older workaround still works: a throwaway "intake"
as
whose
is never invoked (no transition routes back to it) — but a
is cheaper (no model config, no plugin) and is what the framework's own smoke tests use. Either way, the same applies to
TransitionGraph.sequence([a, b, c])
:
is
's turn, so if
is supposed to
produce the first artifact (not just relay a prompt), prepend a seeder —
sequence([user, a, b, c])
.
(Symptom when you get this wrong: the second agent in the chain receives the bare kickoff prompt as if it were the first agent's output, "responds" to something that isn't there, and the pipeline produces nonsense — often hallucinating that an artifact exists.)
Built-in targets
| Target | Decision |
|---|
| Hand off to a specific named agent. |
| Advance through the participant order. |
| Same speaker continues (rare; "let me elaborate" patterns). |
RevertToInitiatorTarget()
| Hand back to whoever opened the channel. |
TerminateTarget(reason="…")
| End the channel; reason flows on EV_CHANNEL_CLOSED.event_data["reason"]
|
Built-in conditions
| Condition | Fires when |
|---|
| Every accepted envelope. |
| The just-accepted envelope was sent by this agent. |
| The previous turn called this tool by name (matched via on the packet). |
ContextEquals(key, value)
| Channel-scoped equals . Missing keys compare as . |
For combinations beyond what these compose to, see "Custom targets / conditions" below.
Convenience factories
python
# Pipeline: a → b → c → terminate.
graph = TransitionGraph.sequence([alice.agent_id, bob.agent_id, carol.agent_id])
# Round-robin with a cap: a → b → c → a → b → c → terminate after 6 turns.
graph = TransitionGraph.round_robin(
participants=[alice.agent_id, bob.agent_id, carol.agent_id],
max_turns=6,
)
uses
Always() → RoundRobinTarget()
and a
TerminateTarget("round_robin_complete")
default — but since
matches every turn, the default is unreachable and the channel actually closes with reason
once the cap is hit (set
or it cycles forever).
uses
FromSpeaker(steps[i]) → AgentTarget(steps[i+1])
chains with a
TerminateTarget("sequence_complete")
default —
that one is reachable: after the last step speaks, no
rule matches, so the default fires and you get
.
Conditional handoff (the most common manual graph)
A triage agent inspects each request and routes to a specialist via a tool call:
python
from ag2.network import (
AgentTarget, FromSpeaker, RevertToInitiatorTarget,
TerminateTarget, ToolCalled, Transition, TransitionGraph,
)
graph = TransitionGraph(
initial_speaker=triage.agent_id,
transitions=[
# Triage's `escalate` tool fired → route to security.
Transition(when=ToolCalled("escalate"),
then=AgentTarget(security.agent_id)),
# Once security has spoken, hand back to whoever opened the channel.
Transition(when=FromSpeaker(security.agent_id),
then=RevertToInitiatorTarget()),
# Otherwise triage's reply goes to the generic responder.
Transition(when=FromSpeaker(triage.agent_id),
then=AgentTarget(general.agent_id)),
],
default_target=TerminateTarget(reason="triage_complete"),
max_turns=20,
)
For each
ToolCalled(name) → AgentTarget(agent)
transition, attach an
-decorated function on the speaker's
. The simplest form returns a typed
— the framework reads it from the tool's result and routes:
python
from ag2.network import Handoff
# `.tool` lives on `Agent`. The `AgentClient` returned by `hc.register(...)`
# is the network handle (used for `.open()`, `.send()`, `.agent_id`) and does
# NOT expose `.tool`. So: build the Agent, attach tools, *then* register.
triage_agent = Agent("triage", prompt="…", config=config)
@triage_agent.tool
async def escalate(reason: str = "") -> Handoff:
"""Escalate this ticket to the security reviewer."""
# Handoff.target is the participant's Passport NAME, not a raw agent_id —
# the framework resolves name → id at packet-finalize via the hub's name directory.
return Handoff(target="security", reason=reason)
triage = await triage_hc.register(
triage_agent, Passport(name="triage"), Resume(), attach_plugin=False,
)
The typed
return supersedes the matching
rule when both are configured — useful when the
target depends on runtime state. The graph's
rule remains useful as documentation and as a fallback.
Context variables — the read/write loop
Channel-scoped mutable state lives on
WorkflowState.context_vars: dict[str, Any]
, scoped to one workflow channel and persisted as
envelopes on the WAL.
Writing context (from a tool)
python
from ag2.network import ChannelInject, EV_CONTEXT_SET
from ag2.network.workflow_helpers import set_context
async def set_route(route: str, channel: ChannelInject) -> str:
"""Record the routing decision for this channel."""
if channel is None:
return "no active channel"
await set_context(channel, "route", route)
return f"route set to {route!r}"
set_context(channel, key, value)
is a thin wrapper that emits
with the default broadcast audience (
— visible to every participant; the mutation lands on
WorkflowState.context_vars
after fold). Equivalent raw form:
python
await channel.send(
"",
event_type=EV_CONTEXT_SET,
event_data={"set": {"route": route}},
)
The full event_data shape supports both set and delete:
{"set": {...}, "delete": [...]}
. Either is optional; within one envelope
runs before
.
Reading context (in a transition)
python
Transition(
when=ContextEquals(key="route", value="security"),
then=AgentTarget(security.agent_id),
)
ContextEquals(key, value=None)
fires when the key is unset or explicitly deleted. The mutation is folded
before substantive turn checks, so a
rule on the same fold sees the just-set value.
Reading context (in a tool)
python
from ag2.network import ChannelStateInject
async def increment_counter(channel: ChannelInject, state: ChannelStateInject) -> str:
if state is None or channel is None:
return "no active channel"
current = state.context_vars.get("counter", 0)
await set_context(channel, "counter", current + 1)
return f"counter now {current + 1}"
Initial values
python
channel = await alice.open(
type="workflow",
target=[bob.agent_id, carol.agent_id],
knobs={
"graph": graph.to_dict(),
"context_vars": {"escalation_level": 0, "ticket_id": ticket_id},
},
)
The stuck-routing trap
is
sticky — once a key is set, every subsequent fold re-evaluates it. If you have
ContextEquals("route", "security") → AgentTarget(security)
near the top of the list, you'll bounce back to security forever (or until
). Two fixes:
- List terminate rules before context-conditions:
FromSpeaker(security) → TerminateTarget(...)
- Have the routed agent clear the key: the security agent's tool emits with when done.
Fix #1 is the more common pattern.
Feedback loop — context-driven termination
python
graph = TransitionGraph(
initial_speaker=intake.agent_id,
transitions=[
# Approve sets done=True → terminate.
Transition(when=ContextEquals("done", value=True),
then=TerminateTarget("approved")),
# intake → drafter.
Transition(when=FromSpeaker(intake.agent_id),
then=AgentTarget(drafter.agent_id)),
# Alternate drafter ↔ reviewer.
Transition(when=FromSpeaker(drafter.agent_id),
then=AgentTarget(reviewer.agent_id)),
Transition(when=FromSpeaker(reviewer.agent_id),
then=AgentTarget(drafter.agent_id)),
],
default_target=TerminateTarget("max_iterations"),
max_turns=10,
)
# `reviewer_agent` is the `Agent`; `reviewer` is the `AgentClient` returned
# by `register()` and used in the graph above (for `reviewer.agent_id`).
# Tools must be attached to the Agent before registration.
@reviewer_agent.tool
async def approve(reason: str, channel: ChannelInject) -> str:
"""Mark the draft approved; the graph terminates with reason='approved'."""
if channel is None:
return "no channel"
await set_context(channel, "done", True)
return f"approved: {reason}"
The reviewer's
call writes
; the reviewer's reply text lands on the same packet fold;
ContextEquals("done", True)
matches and the channel terminates with
. Without
, the drafter/reviewer alternation continues until
hits
and the channel closes with reason
. (The
default_target=TerminateTarget("max_iterations")
above only fires if some turn produced
no matching transition — here one of the
rules always matches, so in this graph it's effectively dead; keep it as a sane fallback for graphs where it can be reached.)
The eight cookbook patterns
The AG2 docs ship a
Pattern Cookbook with runnable examples. One-line summaries — each one is a
channel with a specific graph:
| Pattern | Graph | When |
|---|
| Pipeline | TransitionGraph.sequence([a, b, c])
| Linear A→B→C→terminate (research → write → edit). |
| Hierarchical | Coordinator + FromSpeaker(specialist) → RevertToInitiatorTarget()
| Coordinator dispatches; specialist returns to coordinator. |
| Star | Hub queries N spokes; ToolCalled("synthesise") → TerminateTarget
| Multi-source aggregation, parallel data gathering. |
| Escalation | Tiered + ToolCalled("resolve") → TerminateTarget("resolved")
| Support tiers, multi-level approval. |
| Redundant | with N proposers + evaluator | Niche; novelty/ideation with comparison. |
| Feedback loop | ContextEquals("done", True) → TerminateTarget
| Drafter ↔ reviewer until approved. |
| Context-aware routing | Triage sets ; ContextEquals(category, X) → AgentTarget(specialist_X)
| Support triage, skill-based dispatch. |
| Triage with tasks | + initial values | Triage produces plan; sequence executes the plan. |
Pipeline, escalation, context-aware routing, and feedback loop together cover most real-world needs. The cookbook page has runnable end-to-end code for each.
Custom targets and conditions
When the built-ins don't fit, implement the
and register the class so it round-trips through
TransitionGraph.to_dict()
:
python
from typing import ClassVar
from dataclasses import dataclass, field
from ag2.network import (
Envelope,
TransitionDecision,
TransitionTarget,
register_target,
)
@dataclass(slots=True)
class HighestRankedReviewer(TransitionTarget):
name: ClassVar[str] = "highest_ranked_reviewer"
role_priority: list[str] = field(default_factory=list)
def resolve(self, state, envelope: Envelope) -> TransitionDecision:
# ... look up the next reviewer based on domain logic ...
return TransitionDecision(next_speaker=chosen_id)
register_target(HighestRankedReviewer)
Same pattern for conditions — implement
evaluate(state, envelope) -> bool
, set a
, call
register_condition(MyCondition)
. Both customisations persist via
: the
field is the key, dataclass fields become the args dict reconstituted by
.
The packet execution model and idempotent tools
Each
round on a workflow channel commits to the WAL atomically as a single
envelope. The packet carries the agent's routing decision (
matched against
rules, or a pre-resolved
from a typed
), the round's body text, and a reserved
slot.
envelopes from tool calls land
before the packet, so a
rule on the same fold sees the just-set value.
Atomicity has a consequence: if the agent crashes mid-packet, the channel reverts to its pre-packet state and the input is re-dispatched. Tool calls in that packet execute again on retry.
Tools that touch external systems must be idempotent under retry:
- Use the external service's idempotency-key feature where available (Stripe, S3). Derive a stable key from
(channel_id, round_counter, tool_name)
- For database writes, use upsert () rather than blind insert.
- For tools that genuinely cannot be idempotent, gate them behind HITL or run them in single-tool rounds so the packet boundary is tighter.
Migration from classic GroupChat
The translation is mostly mechanical. The two systems share the vocabulary — speakers, conditions, targets, terminations — but the new one is data-first (a
is JSON-serialisable and survives
).
Concept mapping
| AG2 Classic | AG2 network | Notes |
|---|
| channel with participants | The hub plays 's role. |
| + | Turn-taking moves into the hub. |
| (per-channel, not per-agent) | Handoffs described once at channel level. |
| | Takes id, not reference. |
| RevertToInitiatorTarget()
| Same semantics. |
| | Same. |
| TerminateTarget(reason="…")
| Now carries a reason. |
| or custom | Register via . |
| Custom | Same. |
| from a tool | Typed return | Handoff(target=..., reason=...)
|
| Custom | Implement , . |
| on the graph | Same hard cap. |
Side-by-side: round-robin
python
# Classic
groupchat = GroupChat(agents=[alice, bob, carol],
speaker_selection_method="round_robin", max_round=6)
manager = GroupChatManager(groupchat=groupchat, llm_config=llm_config)
alice.initiate_chat(manager, message="Topic: …")
# AG2 network
graph = TransitionGraph.round_robin(
participants=[alice.agent_id, bob.agent_id, carol.agent_id],
max_turns=6,
)
channel = await alice.open(
type="workflow",
target=[bob.agent_id, carol.agent_id],
knobs={"graph": graph.to_dict()},
)
await channel.send("Topic: …")
Side-by-side: conditional handoff ( → )
python
# Classic
@triage.register_for_llm(description="Escalate to security review.")
def escalate(reason: str) -> ReplyResult:
return ReplyResult(target=AgentTarget(security_reviewer), message=...)
# AG2 network — attach to the Agent *before* registering with the hub
@triage_agent.tool
async def escalate(reason: str = "") -> Handoff:
# Handoff.target is the Passport NAME (resolved name → id by the framework),
# whereas AgentTarget in the graph takes the raw agent_id.
return Handoff(target="security_reviewer", reason=reason)
# Plus the matching transition in the graph (as documentation/fallback):
Transition(when=ToolCalled("escalate"), then=AgentTarget(security_reviewer.agent_id))
Migration checklist
- Stand up a hub (
Hub.open(MemoryKnowledgeStore())
- Wrap each existing via with a and .
- Translate config into a . Use / factories where they fit; build manually for conditional logic.
- Replace with
alice.open(type="workflow", target=[...], knobs={"graph": graph.to_dict()})
- Wait for via
wait_for_channel_event(...)
- Inspect via and
hub._audit_log.read_all()
State object
python
@dataclass(slots=True)
class WorkflowState:
participant_order: list[str]
expected_next_speaker: str | None
last_speaker_id: str | None = None
last_envelope_id: str | None = None
turn_count: int = 0
pending_close_reason: str = ""
creator_id: str = ""
graph_data: dict = field(default_factory=dict)
context_vars: dict[str, Any] = field(default_factory=dict)
expected_next_speaker = None
signals "channel should terminate" — the adapter's
reads this and returns
AdapterResult(next_state=CLOSING, ...)
.
is the serialised graph; the adapter rebuilds
on every fold so there's no mutable graph state in memory.
Quick reference — imports
python
from ag2.network import (
# Graph and transitions
TransitionGraph,
Transition,
TransitionTarget,
TransitionCondition,
TransitionDecision,
register_target,
register_condition,
# Built-in targets
AgentTarget,
RoundRobinTarget,
StayTarget,
RevertToInitiatorTarget,
TerminateTarget,
# Built-in conditions
Always,
FromSpeaker,
ToolCalled,
ContextEquals,
# Typed handoff return for tools
Handoff,
# Context variables
EV_CONTEXT_SET,
ChannelInject,
ChannelStateInject,
# Channel types
WORKFLOW_TYPE,
# Seeding a workflow (kickoff participant)
HumanClient, # via HubClient.register_human(Passport(name=..., kind="human"))
)
from ag2.network.workflow_helpers import set_context, delete_context