Distributed Deployment

The network is not limited to one process. Replace LocalLink with WsLink and the hub becomes a server that agents anywhere on the network can connect to over WebSocket — entirely over the wire, no shared memory, no in-process hub reference.

Install#

WsLink and serve_ws require the network-ws extra:

pip install "ag2[network-ws]"

Architecture#

        ┌─────────────── hub server ────────────────┐
        │  Hub + serve_ws  ·  registry · WAL · auth │
        └───▲──────────────────────────▲────────────┘
            │  ws://hub:8765           │  ws://hub:8765
   ┌────────┴──────────┐     ┌─────────┴──────────┐
   │  process A        │     │  process B         │
   │  HubClient(WsLink)│     │  HubClient(WsLink) │
   │  alice            │     │  bob               │
   └───────────────────┘     └────────────────────┘

Every control-plane call (register, open channel, post envelope, WAL read) travels as a RequestFrame / ResponseFrame RPC over the WebSocket. Every inbound notify arrives as a NotifyFrame and is ack'd by a ReceiptFrame. The HubClient API is identical whether you pass a LocalLink or a WsLink — the transport is the only thing that changes.

Starting the Hub Server#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

import asyncio
import contextlib

from ag2.knowledge import MemoryKnowledgeStore
from ag2.network import Hub, serve_ws

async def main() -> None:
    hub = await Hub.open(MemoryKnowledgeStore())
    async with serve_ws(hub, "127.0.0.1", 8765) as server:
        host, port = server.sockets[0].getsockname()[:2]
        print(f"listening on ws://{host}:{port}", flush=True)
        with contextlib.suppress(asyncio.CancelledError):
            await asyncio.Future()  # serve until interrupted
    await hub.close()

asyncio.run(main())

serve_ws(hub, host, port) is an async context manager. It binds a WebSocket server, hands each incoming connection its own WsLinkEndpoint, and lets the hub dispatch from there. Pass port=0 to bind an ephemeral port and read the real one from server.sockets[0].getsockname()[1].

Connecting a Remote Agent#

From any other process — same machine, different container, or across a real network. The example connects to ws://hub-host:8765; reaching a hub from another machine needs it bound to a reachable address with real authentication (see the warning above), while on one machine you can use ws://127.0.0.1:8765:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import asyncio

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.network import HubClient, WsLink

async def main() -> None:
    hub_client = HubClient(WsLink("ws://hub-host:8765"))
    await hub_client.open()  # WebSocket connect + handshake

    agent = Agent("bob", prompt="Answer in one short sentence.", config=AnthropicConfig(model="claude-haiku-4-5"))
    bob = await hub_client.register(agent)

    print(f"registered as {bob.agent_id}; awaiting channels")
    await asyncio.Future()  # stay connected; default handler answers inbound consults

asyncio.run(main())

HubClient(WsLink(url)) puts the client in remote mode: every register, open, send, and read_wal call is an RPC round-trip to the hub. The AgentClient surface (bob.open(...), bob.wait_for_channel_event(...), channel.send(...)) is identical to the in-process API.

Agents backed by different providers can share the same hub and the same channel — the hub is provider-neutral.

Sending a Cross-Process Consult#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

import asyncio

from ag2 import Agent
from ag2.config import OpenAIConfig
from ag2.network import EV_TEXT, HubClient, WsLink
from ag2.network.adapters.consulting import CONSULTING_TYPE

async def main() -> None:
    hub_client = HubClient(WsLink("ws://hub-host:8765"))
    await hub_client.open()

    alice = await hub_client.register(
        Agent("alice", prompt="You are a coordinator.", config=OpenAIConfig(model="gpt-4o-mini")),
    )

    # open() and send() travel to the hub over the wire;
    # the hub invites bob over bob's own WebSocket connection.
    channel = await alice.open(type=CONSULTING_TYPE, target=["bob"])
    await channel.send("What is 12 times 11? Reply with just the integer.")

    reply = await alice.wait_for_channel_event(
        channel_id=channel.channel_id,
        predicate=lambda e: e.event_type == EV_TEXT and e.sender_id != alice.agent_id,
        timeout=90.0,
    )
    print(reply.event_data["text"])  # 132
    await hub_client.close()

asyncio.run(main())

The channel invite travels hub → bob over bob's WebSocket. Bob's default handler answers, and the reply comes back hub → alice over alice's WebSocket. The hub brokers the exchange; neither agent holds a reference to the other.

All four channel adapters work cross-process without change: consulting, conversation, discussion, and workflow.

Authentication#

For deployments where agents must authenticate at the WebSocket handshake, build the hub with an AuthRegistry:

1
2
3
4

from ag2.network import ApiKeyAuth, AuthRegistry, Hub

auth = AuthRegistry([ApiKeyAuth(keys={"token-alice", "token-bob"})])
hub = await Hub.open(store, auth=auth)

Each agent passes its token inside Passport.auth:

1
2
3
4
5
6
7

from ag2.network import AuthBlock, Passport

passport = Passport(
    name="bob",
    auth=AuthBlock(scheme="api_key", claim={"token": "token-bob"}),
)
bob = await hub_client.register(agent, passport)

The hub's AuthRegistry validates the claim before binding the connection. Raise AuthError from a custom AuthAdapter (a Protocol) to reject any scheme you define.

What a connection may do#

Every request on a connection runs as the agents bound to it — the ones it registered, or re-attached to with attach. One connection may hold several agents. The hub rejects, with AccessDeniedError, a request that:

• changes another agent's resume, skill, or rule, unregisters it, or acts on its behalf (create_channel, pending_turns_for, …);
• posts an envelope whose sender_id is not bound to the connection;
• reads or closes a channel none of the connection's agents participates in;
• updates, checkpoints, or reads the checkpoint of a task another agent owns — for a checkpoint the hub has no task record for, the owner is whichever connection's agents wrote it first over the wire. A checkpoint written in-process (or by an older hub) has no recorded writer, so it is readable by any connection and claimed by the first wire writer.

Independently, a channel accepts events only from its participants — except a task cancel request (as the tasks tool's cancel action sends it) addressed to the owner of a live task in that channel. A non-participant may send at most one cancel request per task; a retry gets ProtocolError — and attach must present the auth scheme the identity registered with. Registering a kind="remote_agent" identity over the wire is off unless the hub is opened with allow_remote_agent_registration=True; then it needs a connection that already holds one of your agents; those agents own it, and only a connection holding an owner can re-attach it. Enable it only if you trust your clients with naming: remote-agent registration skips authentication, so any admitted client can take any unused name first, and the auth.scheme / auth.claim it registers become the routing data your RemoteAgentProxy dispatches with — the client chooses where the proxy sends that agent's traffic. Task ids must be a single path segment (letters, digits, ., _, -). Agent names may not equal, or look like, an agent id (32 lowercase hex characters), and get_agent resolves an exact id before a name. Over the wire, fire_task_event accepts only the task mirror's mirror_failed report — its owner_id must be one of the connection's agents and its channel_id the task's own channel (or, for a task the hub hasn't observed, a channel that agent is in); terminal task events come from the hub itself when update_task finishes a task. Names that only resemble an id or another name (upper-case hex, dashed UUIDs, look-alike characters) are accepted, so show ids alongside names wherever identity matters.

1
2
3
4
5
6

from ag2.network import AccessDeniedError

try:
    await mallory_client.set_skill(bob_agent_id, "...")
except AccessDeniedError:
    ...  # "connection is not bound to agent '...'"

list_channels() without an agent_id lists only the channels of the connection's own agents. Discovery reads — get_agent, list_agents, resumes, skills, rules — stay open to every connection. get_task and list_tasks show a task only to its owner and to the participants of its channel. An envelope's audience may name only participants of its channel. Once attach re-binds an agent to a new connection, the old connection can no longer act as it.

In-process clients (LocalLink with a hub reference) call the hub directly and are not subject to these checks.

At-Least-Once Delivery#

The hub guarantees each envelope is delivered at least once across reconnects:

• Each AgentClient maintains an inbox cursor per channel — the envelope_id of the last successfully processed envelope.
• Inbound envelopes are ack'd with ReceiptFrame; a nack causes immediate replay.
• On reconnect, pass since_envelope_id to replay any unacked envelopes from that point:

1
2
3
4

hub_client = HubClient(WsLink(url))
await hub_client.open()
bob = await hub_client.attach(agent, name="bob", passport=bob_passport, since_envelope_id=last_acked_id)
await hub_client.resume_pending_turns(bob)

attach re-binds an existing identity to a fresh connection. Pass the identity's own passport (with its AuthBlock credential): the hub never returns another agent's — or your own — auth claim, so it cannot be recovered from get_agent. Keep the passport you registered with: the AgentClient.passport a remote register returns has no claim and cannot re-attach a credentialed identity. resume_pending_turns re-fires any turns the protocol still expects from this agent. The default notify handler is idempotent under redelivery — causation-id deduplication short-circuits duplicate model turns without double-posting.

Task Durability#

Tasks can be checkpointed through the hub so state survives a process restart:

1
2
3
4
5
6
7
8
9

from ag2.network import HubBackedCheckpointStore

checkpoint_store = HubBackedCheckpointStore(hub_client)

# inside a tool or agent turn:
await task.checkpoint({"step": 3, "partial_result": "..."})

# on another node, after a restart:
recovered = await agent.resume_from(task_id, checkpoint_store)

HubBackedCheckpointStore satisfies the CheckpointStore Protocol by delegating writes and reads to the hub's KnowledgeStore. Pass a Hub for in-process durability or a HubClient for cross-process. For checkpoints that survive a hub restart, use DiskKnowledgeStore(path) on the hub.

Production Notes#

Where to Next#

• Hub & Identity — AuthRegistry, ApiKeyAuth, governance rules.
• Agent Clients — custom envelope handlers, replacing the default handler.
• Governance, Audit & Observability — per-channel expectations and the hub audit log.