undefined

The default handler wraps the LLM's plain-text reply into the workflow `EV_PACKET` either way. The examples below pass `attach_plugin=False` because they're pure pipeline workers; drop it if your agents need `delegate` / `peers` / etc. Full detail: `ag2-network-tools-and-views` → "Adapter-owned tools (`tools_for`)".

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())

@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

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

`register_human(passport, *, resume=None, rule=None, auto_ack_invites=True)` returns a `HumanClient` (`HubClient.register(...)` now *rejects* `kind="human"` and points you here). It auto-acks invites so the quorum handshake completes without UI round-trips; pass `auto_ack_invites=False` to gate joins by hand. Drive it with `on_envelope(callback)` (push) or `next_envelope(...)` / `envelopes()` (pull) — see `ag2-network-quickstart` → "`HumanClient`".

If you'd rather not introduce a `HumanClient`, the older workaround still works: a throwaway "intake" `Agent` as `initial_speaker` whose `Agent.ask` is never invoked (no transition routes back to it) — but a `HumanClient` 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])`: `channel.send(...)` is `a`'s turn, so if `a` 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.)

undefined

`round_robin` uses `Always() → RoundRobinTarget()` and a `TerminateTarget("round_robin_complete")` default — but since `Always()` matches every turn, the default is unreachable and the channel actually closes with reason `"max_turns"` once the cap is hit (set `max_turns` or it cycles forever). `sequence` 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 `FromSpeaker` rule matches, so the default fires and you get `"sequence_complete"`.

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,
)

from ag2.network import Handoff

The typed `Handoff` return supersedes the matching `ToolCalled` rule when both are configured — useful when the *target* depends on runtime state. The graph's `ToolCalled` rule remains useful as documentation and as a fallback.

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}"

await channel.send(
    "",
    event_type=EV_CONTEXT_SET,
    event_data={"set": {"route": route}},
)

Transition(
    when=ContextEquals(key="route", value="security"),
    then=AgentTarget(security.agent_id),
)

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}"

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},
    },
)

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,
)

The reviewer's `approve` call writes `done=True`; the reviewer's reply text lands on the same packet fold; `ContextEquals("done", True)` matches and the channel terminates with `"approved"`. Without `approve`, the drafter/reviewer alternation continues until `turn_count` hits `max_turns` and the channel closes with reason `"max_turns"`. (The `default_target=TerminateTarget("max_iterations")` above only fires if some turn produced *no* matching transition — here one of the `FromSpeaker` rules always matches, so in this graph it's effectively dead; keep it as a sane fallback for graphs where it can be reached.)

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)

undefined

undefined

undefined

undefined

@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)

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