Agent Clients

The agent-side of the network. A HubClient represents one process's connection to the hub; it produces AgentClients — one per registered Agent. Each AgentClient carries a notify handler that decides what the agent does when an envelope arrives.

HubClient — one per process#

1
2
3
4

from autogen.beta.network import HubClient, LocalLink

link = LocalLink(hub)            # transport factory bound to this hub
hc = HubClient(link, hub=hub)

A HubClient is the per-process registration broker. In a single-process script you may construct one per agent (each gets its own duplex link); in a real deployment you'd typically have one HubClient per process and register all of that process's agents on it.

The probe and view-policy methods exist so custom handlers (next section) don't have to reach into hub internals.

Registering an Agent#

1
2
3
4
5
6
7
8

agent_client = await hc.register(
    agent,                     # autogen.beta.Agent
    passport,                  # autogen.beta.network.Passport
    resume,                    # autogen.beta.network.Resume
    skill_md=None,             # optional — markdown describing this agent's skill
    rule=None,                 # optional — Rule(...) for governance
    attach_plugin=True,        # whether to install the default notify handler
)

attach_plugin=True (the default) installs the default_handler on this client. Pass False if you want full control over inbound envelope handling — typical for headless workers, gateways, or custom orchestration logic.

AgentClient#

channel = await agent_client.open(...) returns a Channel handle scoped to one channel id. Use channel.send(text, audience=...) for ordinary text sends, channel.close() for explicit termination, channel.info() for the latest metadata.

The Default Handler#

When you register with attach_plugin=True, the client installs default_handler. It routes inbound envelopes:

The handler wraps the entire turn path (WAL slice, view projection, extract_turn_input, agent.ask, round-envelope build, outbound send) in a single trap: a crash is routed through HubClient.report_turn_failure → Hub.report_turn_failure, which fans on_turn_failed out to every HubListener (the built-in AuditLog records AUDIT_KIND_TURN_FAILED). No reply is posted, but the channel stays active and the next envelope flows normally — see Turn-failure resilience.

The handler is decomposed into public hooks so you can override only the parts you care about:

1
2
3
4
5

from autogen.beta.network import (
    read_wal_until,
    resolve_view_policy,
    stamp_dependencies,
)

Custom Handlers#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from autogen.beta.network import Envelope, EV_TEXT

async def gateway_handler(envelope: Envelope) -> None:
    if envelope.event_type != EV_TEXT:
        return
    # forward to your own external system instead of running an LLM
    text = envelope.event_data.get("text", "")
    await my_external_queue.put({"from": envelope.sender_id, "text": text})

agent_client.on_envelope(gateway_handler)

Common patterns:

• Headless worker — register with attach_plugin=False and install a handler that pulls work directly off the hub without running an LLM.
• Selective override — install a handler that handles only one event type (e.g. custom invite policy) and falls back to default_handler for everything else.
• Filtered forwarding — wrap default_handler with pre/post hooks for logging, rate limiting, or routing.

Transport — LocalLink#

1
2
3
4

from autogen.beta.network import LocalLink, LinkClient, LinkEndpoint

link = LocalLink(hub)
client_link = link.client()  # produces a fresh LocalLinkClient bound to a fresh LocalLinkEndpoint

LocalLink is the in-process transport. Each HubClient(link, hub=hub) lazily creates one LocalLinkClient/LocalLinkEndpoint pair on first use:

• LocalLinkClient — the agent-process side; sends frames toward the hub, receives notify frames from it.
• LocalLinkEndpoint — the hub-process side; the inverse.

Both sides exchange Frame records via async queues. Frame types are re-exported at the package level: HelloFrame, WelcomeFrame, SendFrame, ReceiptFrame, NotifyFrame, AcceptFrame, ErrorFrame, PingFrame, PongFrame, SubscribeFrame, UnsubscribeFrame.

The transport layer is a Protocol:

class LinkClient(Protocol):
    async def open(self) -> None: ...
    async def send_frame(self, frame: Frame) -> None: ...
    def frames(self) -> AsyncIterator[Frame]: ...
    async def close(self) -> None: ...

Cross-process or cross-host transports plug in here. LocalLink is the only built-in, but the abstraction allows for future Redis/WebSocket/gRPC implementations without changing any client code.

Inbox & Backpressure#

Every AgentClient has an inbox bounded by its Rule.inbox.max_pending (default unbounded). When the inbox fills, sends to that agent fail with InboxFull. The wait_for_channel_event and the default handler drain the inbox in order; custom handlers should do the same — never block forever in a callback.

Rule.inbox.high_water is a soft threshold below the hard cap: when a dispatch first pushes a recipient over it, the hub fires on_inbox_pressure(agent_id, pending, cap) on every listener (and on a Hub subclass override) — once per crossing, not on every subsequent envelope. It defaults to None, which resolves to 80% of max_pending; set 0 to disable the signal. See HubListener.

Closing Down#

1
2
3

await alice_hc.close()
await bob_hc.close()
await hub.close()

HubClient.close() cancels the link's listening task and unsubscribes all clients. Always pair with the matching register calls; otherwise the hub keeps the registration in its registry.