Ag2 Network Tools And Views

AG2 Network — Tools, Views & Custom Handlers

Everything on the agent/client side of the network — the mirror image of ag2-network-governance (which is hub-side). This covers what the LLM sees of the channel (views), which actions it can take (the five auto-injected plugin tools plus the adapter-owned say), what other agents know about it (skill markdown), how to replace the handler entirely, and the full Envelope reference.

Prerequisite: read ag2-network-quickstart first. This skill assumes you know Hub.open, HubClient.register, the channel lifecycle, and basic agent_client.open(...) / channel.send(...).

When to use

• "Limit / extend the LLM's network tool surface" (plugin tools vs. adapter tools_for)
• "Write a custom envelope handler"
• "Build a gateway / headless worker that doesn't run an LLM"
• "Add a non-LLM participant — a person at a UI, a bridge, a scripted seeder" (HumanClient)
• "Customise what each agent sees of the channel history (view policy)"
• "Strip / redact / filter envelopes before they reach the LLM"
• "Wire peer discovery via skill markdown"
• "Send a custom event type (myapp.review_request, …)"
• "I need the EV_* constants list / Envelope shape / audience semantics"

Network tools — plugin tools vs. adapter tools

An agent's per-turn LLM tool list is assembled from two streams:

1. Plugin tools — when you register with the default attach_plugin=True, NetworkPlugin adds five identity-level, channel-agnostic verbs to agent.tools: peers / channels / tasks / context / delegate. Same behaviour in any channel.
2. Adapter tools — the channel's adapter offers channel-specific tools per turn via adapter.tools_for(client, metadata, state, participant_id); the default handler resolves them and merges them into agent.ask(tools=...) (cached per (adapter, client) so the schema build cost is paid once). The only built-in adapter tool is say.

So say is not a plugin tool — it's offered by the adapter, gated by turn state:

That's why attach_plugin=False no longer controls whether the LLM sees say — it controls the five plugin verbs only. (See ag2-network-quickstart → "Plugin tools vs. adapter tools" for the user-facing summary.)

delegate and say

# The LLM emits, e.g.:
delegate(target="bob", prompt="What's the right way to model X?", capability="modeling")
say(content="Here's my answer: …")

The framework resolves ChannelInject (current channel), ChannelStateInject, and AgentClientInject automatically inside the notify handler — the LLM never sees those parameters.

Four grouped action-dispatch tools

Each takes an action literal plus action-specific args, keeping the LLM's tool list short.

peers(action) — discovery

channels(action) — lifecycle

tasks(action) — task lifecycle

Two halves: active actions (the agent is inside its own agent.task(...) block) and observation actions (any task the hub has seen).

"start" is intentionally not a tool — calling it from the LLM would bypass the async with agent.task(...) lifecycle that scopes TaskInject correctly. Owners start tasks in their own code; the LLM uses progress / complete once a task is active, and delegate for one-shot remote work.

context(action) — past content

scope="knowledge" reaches into the calling agent's own KnowledgeStore (substring search only — for vector / semantic search, the agent's own loop calls framework-core recall directly).

Adapter-owned tools (tools_for) — and the say double-send

The default handler resolves adapter.tools_for(client, metadata, state, participant_id) each turn and merges the result into agent.ask(tools=...) (alongside the agent's own @tools and any plugin tools). Adapters that take no LLM input — workflow — return []. The others return say, gated by turn state (table above). An adapter could return richer tools too; say is just the only built-in.

Implication for workflow: a workflow agent never sees say — its only path to "say something" is the round-end EV_PACKET the handler builds from reply.body, plus whatever @tool handoff functions you wrote. The old "approve() then say(...) races the round-end EV_PACKET" failure mode is structurally impossible now.

For consulting / conversation / discussion, say is offered — but you rarely need it. The default handler already posts the round-end envelope (build_round_envelope → EV_TEXT(reply.body), or None if empty), so an agent that just replies with text has already spoken. say is for posting an extra message in a turn, or posting into a different channel the agent participates in (channel_id=).

The hazard: an agent that calls say(content="…") and then also returns a non-empty reply body emits two substantive envelopes in one turn. On conversation that's harmless. On consulting the second one trips the strict 1Q1R adapter — ProtocolError: channel '<id>' is closed. Mitigations, most robust first:

• Don't prompt the agent to call say. Its plain reply is the canonical channel reply; the double-send only happens if you explicitly steer it toward say.
• Replace the default handler with one that doesn't also send a round-end envelope — then say is the agent's voice (gateway / headless-worker patterns below).
• attach_plugin=False does not help here — say is adapter-owned, not plugin-owned. It only drops delegate / peers / channels / tasks / context. Use it for a bare agent, not as a say suppressor.

Non-LLM participants — HumanClient

HumanClient is the framework's first-class "participant that isn't an Agent": no LLM, no NetworkPlugin, no assembly policies. A person at a UI, a bridge to another system, a scripted "user" that seeds a workflow — all of these are a HumanClient. Register with register_human (not register, which now rejects kind="human" and points you here):

from ag2.network import HumanClient, Passport

user = await hc.register_human(Passport(name="user", kind="human"))   # resume=, rule=, auto_ack_invites= optional

It satisfies the NetworkClient Protocol — same outbound surface as AgentClient:

ch = await user.open(type="consulting", target="analyst")     # initiate a channel
await user.send(ch.channel_id, "What's our Q3 exposure?")     # EV_TEXT convenience
await user.post_envelope(env)                                 # escape hatch — adapter-shaped envelopes (e.g. workflow EV_PACKET via adapter.build_packet_envelope)

…plus two ways to consume inbound envelopes (an AgentClient only has the notify-handler callback; HumanClient adds an explicit queue):

auto_ack_invites=True (default) makes the human auto-accept channel invites so the hub's quorum handshake completes without a UI round-trip; pass auto_ack_invites=False to gate joins by hand (you'll then post_envelope an EV_CHANNEL_INVITE_ACK yourself). hub.list_agents(kind="human") discovers humans; hub.list_agents(kind="agent") / kind="remote_agent" filter the others.

Typical roles: the kickoff seeder for a workflow channel (FromSpeaker(user) → AgentTarget(first_agent) — see ag2-network-workflow), the human leg of a consulting Q&A, a participant in a discussion round-robin, or a WebSocket/CLI bridge in front of any of those. For a headless agent (an AgentClient that shouldn't run an LLM but still wants the plugin tools / tools_for resolution) you replace its notify handler instead — next.

Replacing the default handler

The default handler does all the "agent receives envelope → auto-ack invites → run LLM (with adapter.tools_for(...) merged in) → post round-end envelope" wiring. Replace it for headless workers, gateways, or any agent that shouldn't run an LLM. (For a participant that was never meant to have an LLM at all, reach for HumanClient above instead of an AgentClient with a custom handler.)

Opting out of the plugin (and/or the default handler)

worker = await hc.register(agent, passport, resume, attach_plugin=False)  # no peers/channels/tasks/context/delegate
worker.on_envelope(my_custom_handler)                                     # ← this is what swaps the handler

These are two independent knobs:

• attach_plugin=False skips NetworkPlugin.register(agent) — the agent's tool list loses peers / channels / tasks / context / delegate (and the NetworkContextPolicy prefix). It does not touch the notify handler, and it does not remove say (that's adapter-owned — see above).
• The default notify handler is active regardless (it's wired by AgentClient, not the plugin — unless you registered with attach_default_handler=False). Call client.on_envelope(callback) to replace it; call client.on_envelope(client._run_default_handler) to restore it.

What you lose when you call client.on_envelope(callback):

The cheap way to keep most of that for free: handle the events you care about yourself, delegate the rest to default_handler.

Handler signature. A callback passed to client.on_envelope(...) is called with just (envelope,) — the framework does await self._on_envelope(envelope). Any client reference inside the handler is captured from the enclosing scope (a closure variable). The exported default_handler, by contrast, has signature (envelope, client) — pass both when delegating.

A gateway handler

from ag2.network import Envelope, EV_TEXT, default_handler


async def gateway_handler(envelope: Envelope) -> None:
    """Forward inbound text to an external system instead of running an LLM."""
    if envelope.event_type != EV_TEXT:
        await default_handler(envelope, client)   # auto-ack invites etc. (client from closure)
        return
    text = envelope.event_data.get("text", "")
    await my_external_queue.put({
        "from": envelope.sender_id,
        "text": text,
        "channel": envelope.channel_id,
    })


client.on_envelope(gateway_handler)

Selective override (fall back to default)

from ag2.network import default_handler, EV_CHANNEL_INVITE


async def selective_handler(envelope: Envelope) -> None:
    if envelope.event_type == EV_CHANNEL_INVITE and envelope.sender_id not in TRUSTED_AGENTS:
        return   # untrusted invite → silent drop; hub will time out and close
    await default_handler(envelope, client)   # default for everything else


client.on_envelope(selective_handler)

Filtered forwarding (pre/post hooks)

async def logged_handler(envelope: Envelope) -> None:
    log.info("inbound %s from %s", envelope.event_type, envelope.sender_id)
    try:
        await default_handler(envelope, client)
    finally:
        log.info("processed %s", envelope.envelope_id)

Bypassing adapter tools

Adapters offer per-turn tools via tools_for(...) (notably say on consulting / conversation / discussion); the default handler merges them into agent.ask(tools=...). If a capable model insists on calling say unprompted (Claude often will) and the resulting double-send is breaking your channel, swap in a custom handler that runs the same substantive path but omits tools= from agent.ask — the agent's static agent.tools (your @tools + any plugin tools) remain, since ask(tools=…) is additional tools, not a replacement. The shape:

async def no_say_handler(envelope, client):
    if envelope.event_type != EV_TEXT:
        await default_handler(envelope, client)         # invite-ack + lifecycle for free
        return
    if not await client._hub_client.can_send(envelope.channel_id, client.agent_id):
        return                                          # not our turn / channel closing

    # ... read_wal_until / resolve_view_policy / stamp_dependencies (hooks above) ...
    reply = await client.agent.ask(current_input, stream=stream, dependencies=deps)   # ← no tools=
    out = adapter.build_round_envelope(metadata=meta, sender_id=client.agent_id, reply=reply,
                                       events=events, state=state, hub=client._hub)
    if out is not None:
        out.causation_id = envelope.envelope_id
        await client.send_envelope(out)

That's the default handler's substantive path minus one line. Same trick works to add tools (pass your own tools=[...]) or to swap the view, the dependencies, the round-envelope shape, etc.

Hooks for selective override

If you want to partially replace the default handler's logic, the handler is decomposed into public hooks:

from ag2.network import (
    read_wal_until,
    resolve_view_policy,
    stamp_dependencies,
)

Use these when your custom handler wants the standard pre-LLM wiring but a custom post-LLM behaviour (or vice versa).

Views — what each LLM sees

ViewPolicy is the projection layer between the channel's WAL and the LLM's history:

class ViewPolicy(Protocol):
    name: str
    async def project(
        self,
        wal: list[Envelope],
        *,
        participant_id: str,
        channel: ChannelMetadata,
        render_envelope: EnvelopeRenderer,           # required — supplied by the adapter
        name_for: NameResolver = default_name_resolver,  # optional — resolves agent_id → Passport.name
    ) -> list[BaseEvent]: ...

It takes the WAL slice this participant should see and returns a list of BaseEvents that the framework feeds into the LLM turn as pre-populated stream history. render_envelope (an EnvelopeRenderer = Callable[[Envelope], str | None]) is supplied per-call by the channel's adapter so views stay adapter-neutral; name_for (a NameResolver = Callable[[str], str]) labels projection lines and defaults to default_name_resolver (identity). Adapters declare a default; you can override per-channel.

Built-in views

All honour audience — an envelope addressed only to [bob] doesn't appear in carol's projection.

from ag2.network import FullTranscript, WindowedSummary, NamedWindowedSummary

view = WindowedSummary(recent_n=12)
projected = await view.project(
    wal_slice,
    participant_id=carol.agent_id,
    channel=metadata,
    render_envelope=adapter.render_envelope,   # required
    name_for=hub.name_for,                      # optional; defaults to identity
)

Resolving the default

from ag2.network import resolve_view_policy

policy = resolve_view_policy(client, metadata)

Reads the adapter manifest's default_view_policy and instantiates the matching view from the registry. The default handler calls this once per turn — custom handlers should too unless they're deliberately bypassing the standard projection model.

Custom views

Implement the protocol, give it a unique name, and use it:

from ag2.events import BaseEvent, ModelMessage, ModelRequest, TextInput
from ag2.network import EV_TEXT, ViewPolicy


class FromOneOnly(ViewPolicy):
    """Show only envelopes from a single named sender."""
    name = "from_one_only"

    def __init__(self, sender_id: str) -> None:
        self.sender_id = sender_id

    async def project(
        self,
        wal,
        *,
        participant_id,
        channel,
        render_envelope,                 # required by the protocol
        name_for=lambda agent_id: agent_id,
    ):
        out: list[BaseEvent] = []
        for env in wal:
            if env.event_type != EV_TEXT or env.sender_id != self.sender_id:
                continue
            text = render_envelope(env)   # adapter renders the envelope to a string (or None)
            if text is None:
                continue
            out.append(
                ModelMessage(text) if env.sender_id == participant_id
                else ModelRequest([TextInput(text)])
            )
        return out

The BaseEvent types you emit shape how the LLM sees the history: ModelRequest([TextInput(...)]) for "from the user," ModelMessage(...) for "from the assistant," etc. Render the envelope body via the supplied render_envelope callback (so the view stays adapter-neutral) rather than reaching into event_data directly. See ag2.events for the full taxonomy.

Picking a view

• Short, focused exchanges → FullTranscript(). Token budget isn't the bottleneck; coherence is.
• Long-running 2-party exchanges → WindowedSummary(recent_n=N) with N tuned to turn density.
• Long-running N-party discussions / workflows → NamedWindowedSummary(recent_n=N) so speaker labels survive the window (this is already the discussion / workflow default).
• Specialist agents that should ignore unrelated chatter → custom view that filters by audience or tags.
• Privacy-sensitive workflows → custom view that strips fields or redacts before projection.

Switching the view doesn't affect the WAL — every envelope is still there, every operator can still inspect it. Only the LLM's perception of history is shaped.

Skills — how an agent describes itself

Skills are markdown-with-frontmatter that the hub stores verbatim and surfaces to peers during peers(action="describe", name=...) lookup. Pass at registration:

agent_client = await hc.register(
    agent,
    Passport(name="researcher"),
    Resume(claimed_capabilities=["research"]),
    skill_md="""\
---
title: Research Assistant
expertise: [policy, finance]
---

# Researcher

A senior policy analyst. Best at:

- Scenario synthesis from multi-source briefs.
- Rebuttal review with confidence scores.

Limitations: not for code review or numerical analysis.
""",
)

Parsing the frontmatter

from ag2.network import parse_skill_frontmatter, ParsedSkill

parsed: ParsedSkill = parse_skill_frontmatter(skill_md)
print(parsed.frontmatter)  # {"title": "Research Assistant", "expertise": [...]}
print(parsed.body)         # the markdown body

Fallback skills

When no skill_md is provided, the hub generates one from the resume so peer lookup doesn't return empty handles:

from ag2.network import render_fallback_skill

skill_md = render_fallback_skill(passport, resume)

Useful when constructing skills programmatically — e.g. a tenant uploads a resume but no markdown.

Updating after registration

await hub.set_skill(agent_id, new_skill_md)

Emits AUDIT_KIND_SKILL_SET. Same audit shape as set_resume; tenant code can replace skills at any time.

The Envelope wire format

@dataclass(slots=True)
class Envelope:
    channel_id: str
    sender_id: str                       # agent_id
    audience: list[str] | None           # None = broadcast to all participants
    event_type: str                      # "ag2.msg.text", "ag2.channel.invite", etc.
    event_data: dict[str, Any]           # event-specific payload

    envelope_id: str = ""                # hub-stamped on accept
    task_id: str | None = None
    causation_id: str | None = None      # envelope_id this one is "in reply to"
    trace_id: str | None = None
    priority: Priority = "normal"        # Priority is a Literal, not an enum (see below)
    depth: int = 0                       # delegation hop count; hub auto-increments on reply path
    idempotency_key: str | None = None   # reserved for cross-process transports
    created_at: str = ""                 # hub-stamped ISO-Z on accept
    ttl_seconds: int | None = None       # per-envelope TTL; None defers to channel expires_at

The hub stamps envelope_id and created_at at admission time (and auto-increments depth on the reply path). Everything else comes from the sender. (There is no sequence field — WAL order is admission order; see "Reading the WAL".)

Substantive events

EV_TEXT carries plain text. EV_PACKET is the workflow adapter's atomic round-end capture — routing decision (matched against ToolCalled rules), accumulated context_vars mutations, and final body text bundled into one envelope. Posted by the framework after each Agent.ask round on a workflow channel; tool authors don't construct these directly.

Channel lifecycle events

Context and task events

Task lifecycle (started / progress / completed / failed / expired) is not carried as wire envelopes — it's mirrored from the owner's stream straight into the hub's TaskMetadata cache by TaskMirror. The two EV_TASK_* constants above are the exception: cancellation has a peer-driven side that needs an addressable envelope. The default handler no-ops on any ag2.task.* event_type.

Audience and visibility

audience: list[str] | None controls who sees the envelope:

• None — broadcast to all participants.
• [agent_id_1, agent_id_2, ...] — only those participants see it.

from ag2.network import visible_to

if visible_to(env, my_agent_id):
    process(env)

Views (FullTranscript, WindowedSummary) honour audience — an envelope addressed only to [bob] doesn't appear in carol's projection.

Priority

Priority = Literal["background", "normal", "urgent"]

Priority is a string literal type, not an enum — set priority="urgent" (etc.) on the envelope directly. Higher-priority envelopes process ahead of lower in queue order. Use sparingly; most application envelopes leave priority at the default "normal".

Causation

causation_id marks an envelope as "in reply to" another:

await channel.send(reply_text, causation_id=incoming_envelope.envelope_id)

The default handler does this automatically when replying to an inbound EV_TEXT. Custom handlers should set it for logical replies — useful for threaded views.

Sending raw envelopes (custom event types)

channel.send(text, audience=...) wraps EV_TEXT for you. For custom event types build an Envelope and post it directly:

from ag2.network import Envelope

envelope = Envelope(
    channel_id=channel.channel_id,
    sender_id=alice.agent_id,
    audience=[bob.agent_id],
    event_type="myapp.review_request",
    event_data={"document_id": "doc-123", "kind": "security"},
)
await alice.send_envelope(envelope)

The hub doesn't validate event_type against any allowlist — custom types pass through unmodified. Adapters fold only the event types they recognise (substantive ones, plus EV_PACKET under workflow, plus lifecycle ones); custom event types land on the WAL and are delivered to the audience but don't advance turn-taking state.

Custom event type guidelines

1. Use a dotted namespace prefix ("myapp.review_request", not "review") to avoid collision with future ag2.* events.
2. Keep event_data JSON-serialisable (no datetimes, dataclasses, etc.) so it round-trips through the store cleanly.
3. If multiple participants should react, set audience=None. If only one, address it specifically; views filter out non-recipients.
4. Don't rely on adapters to do anything with custom types — they pass through. Your custom handler is responsible for processing them.

Reading the WAL

wal = await hub.read_wal(channel.channel_id)
for i, env in enumerate(wal):
    print(f"{i:>3}  {env.event_type}  from={env.sender_id[:8]}")

Envelopes appear in admission order (WAL list order — there is no per-envelope sequence counter). The WAL is the canonical replay surface — Hub.hydrate() re-folds it through each adapter to rebuild in-memory state on restart.

Quick reference — imports

from ag2.network import (
    # Default handler + hooks (for selective override)
    default_handler,
    read_wal_until,
    resolve_view_policy,
    stamp_dependencies,
    # Plugin (auto-attached by attach_plugin=True) — peers/channels/tasks/context/delegate.
    # `say` is NOT here; it comes from the channel adapter's tools_for(...).
    NetworkPlugin,
    # Non-LLM participant
    HumanClient,           # via HubClient.register_human(Passport(name=..., kind="human"))
    PassportKind,          # Literal["agent", "human", "remote_agent"] | None
    # Views
    ViewPolicy,
    FullTranscript,
    WindowedSummary,
    NamedTranscript,
    NamedWindowedSummary,
    # Skills (peer discovery)
    ParsedSkill,
    parse_skill_frontmatter,
    render_fallback_skill,
    # Envelopes + events
    Envelope,
    Priority,
    visible_to,
    EV_TEXT,
    EV_PACKET,
    EV_CHANNEL_INVITE,
    EV_CHANNEL_INVITE_ACK,
    EV_CHANNEL_INVITE_REJECT,
    EV_CHANNEL_OPENED,
    EV_CHANNEL_CLOSED,
    EV_CHANNEL_EXPIRED,
    EV_EXPECTATION_VIOLATED,
    EV_CONTEXT_SET,
    EV_TASK_CANCEL_REQUEST,
    EV_TASK_CANCELLED,
    # Dependency injection keys (for tests outside the notify handler)
    AGENT_CLIENT_DEP,
    CHANNEL_DEP,
    CHANNEL_STATE_DEP,
    HUB_DEP,
    TASK_DEP,
    # Injects (for tool signatures)
    AgentClientInject,
    ChannelInject,
    ChannelStateInject,
    HubInject,
    TaskInject,
)