Quick Start - AG2

Quick Start

The smallest possible end-to-end network scenario: one in-process hub, two agents, a consulting channel that auto-closes after a single Q-and-A.

 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 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65  import asyncio from autogen.beta import Agent from autogen.beta.config import AnthropicConfig from autogen.beta.knowledge import MemoryKnowledgeStore from autogen.beta.network import ( EV_CHANNEL_CLOSED, EV_TEXT, Hub, HubClient, LocalLink, Passport, Resume, ) async def main() -> None: config = AnthropicConfig(model="claude-sonnet-4-6") # Hub: registry + WAL + audit log + adapters live here. hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0) link = LocalLink(hub) # in-process duplex transport # Each agent gets its own HubClient (its own duplex pair to the hub). alice_hc = HubClient(link, hub=hub) bob_hc = HubClient(link, hub=hub) alice = await alice_hc.register( Agent("alice", prompt="Ask one focused question and stop.", config=config), Passport(name="alice"), Resume(), ) bob = await bob_hc.register( Agent("bob", prompt="Answer in one short sentence.", config=config), Passport(name="bob"), Resume(), ) # Strict 1Q1R; the adapter auto-closes on bob's reply. channel = await alice.open(type="consulting", target="bob") await channel.send( "What's the single most important property of a distributed system?", audience=[bob.agent_id], ) # Bob's default handler runs Agent.ask on the inbound EV_TEXT, sends the # reply, and ConsultingAdapter posts EV_CHANNEL_CLOSED. close_env = await alice.wait_for_channel_event( channel_id=channel.channel_id, predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED, timeout=60.0, ) print(f"closed: {close_env.event_data.get('reason')!r}") # Replay the conversation from the hub's write-ahead log. wal = await hub.read_wal(channel.channel_id) for env in wal: if env.event_type == EV_TEXT: speaker = "alice" if env.sender_id == alice.agent_id else "bob" print(f"{speaker}: {env.event_data['text']}") await alice_hc.close() await bob_hc.close() await hub.close() asyncio.run(main())


`<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br>22<br>23<br>24<br>25<br>26<br>27<br>28<br>29<br>30<br>31<br>32<br>33<br>34<br>35<br>36<br>37<br>38<br>39<br>40<br>41<br>42<br>43<br>44<br>45<br>46<br>47<br>48<br>49<br>50<br>51<br>52<br>53<br>54<br>55<br>56<br>57<br>58<br>59<br>60<br>61<br>62<br>63<br>64<br>65<br>`	<br>import asyncio<br>from autogen.beta import Agent<br>from autogen.beta.config import AnthropicConfig<br>from autogen.beta.knowledge import MemoryKnowledgeStore<br>from autogen.beta.network import (<br> EV_CHANNEL_CLOSED,<br> EV_TEXT,<br> Hub,<br> HubClient,<br> LocalLink,<br> Passport,<br> Resume,<br>)<br>async def main() -> None:<br> config = AnthropicConfig(model="claude-sonnet-4-6")<br> # Hub: registry + WAL + audit log + adapters live here.<br> hub = await Hub.open(MemoryKnowledgeStore(), ttl_sweep_interval=0)<br> link = LocalLink(hub) # in-process duplex transport<br> # Each agent gets its own HubClient (its own duplex pair to the hub).<br> alice_hc = HubClient(link, hub=hub)<br> bob_hc = HubClient(link, hub=hub)<br> alice = await alice_hc.register(<br> Agent("alice", prompt="Ask one focused question and stop.", config=config),<br> Passport(name="alice"),<br> Resume(),<br> )<br> bob = await bob_hc.register(<br> Agent("bob", prompt="Answer in one short sentence.", config=config),<br> Passport(name="bob"),<br> Resume(),<br> )<br> # Strict 1Q1R; the adapter auto-closes on bob's reply.<br> channel = await alice.open(type="consulting", target="bob")<br> await channel.send(<br> "What's the single most important property of a distributed system?",<br> audience=[bob.agent_id],<br> )<br> # Bob's default handler runs Agent.ask on the inbound EV_TEXT, sends the<br> # reply, and ConsultingAdapter posts EV_CHANNEL_CLOSED.<br> close_env = await alice.wait_for_channel_event(<br> channel_id=channel.channel_id,<br> predicate=lambda e: e.event_type == EV_CHANNEL_CLOSED,<br> timeout=60.0,<br> )<br> print(f"closed: {close_env.event_data.get('reason')!r}")<br> # Replay the conversation from the hub's write-ahead log.<br> wal = await hub.read_wal(channel.channel_id)<br> for env in wal:<br> if env.event_type == EV_TEXT:<br> speaker = "alice" if env.sender_id == alice.agent_id else "bob"<br> print(f"{speaker}: {env.event_data['text']}")<br> await alice_hc.close()<br> await bob_hc.close()<br> await hub.close()<br>asyncio.run(main())<br>

Expected output (Sonnet's exact words will differ on each run):

closed: 'consulting_complete'
alice: What's the single most important property of a distributed system?
bob: Fault tolerance — because a system that can't survive partial failures defeats its entire purpose.

What Just Happened

In order:

Hub.open(MemoryKnowledgeStore()) — boots an in-process hub. The KnowledgeStore is where the hub persists its audit log, registry, and write-ahead logs (here in memory).
LocalLink(hub) — a transport factory. Each HubClient constructed against the same link gets its own duplex queue pair to the hub.
HubClient(link, hub=hub) — one per process boundary. In a real deployment alice and bob would each live in their own process, each with one HubClient. Here they share a process for clarity.
hc.register(agent, passport, resume) — registers an Agent with the hub. Returns an AgentClient whose agent_id is hub-stamped.
alice.open(type="consulting", target="bob") — alice creates a consulting channel with bob as respondent. Internally: hub posts EV_CHANNEL_INVITE to bob → bob's default handler auto-acks → hub posts EV_CHANNEL_OPENED and alice.open(...) returns with channel.state == ACTIVE.
channel.send(text, audience=...) — alice sends an EV_TEXT envelope.
Bob's default handler — receives the EV_TEXT, probes whether the adapter would accept a reply right now (it would — bob hasn't replied yet), runs Agent.ask(text), and sends bob's reply back through bob's own channel handle.
ConsultingAdapter — sees both initiator_sent and respondent_replied are true, returns AdapterResult(next_state=CLOSED, auto_close_reason="consulting_complete"). Hub posts EV_CHANNEL_CLOSED.
alice.wait_for_channel_event(...) — alice's loop wakes when she receives the close envelope.
hub.read_wal(channel_id) — replays every envelope the hub recorded for the channel. Each envelope is hub-stamped (id, timestamp, sender, audience, event_type, event_data).

Mental Hooks

The Hub is the only authoritative state. Every send goes through it; every observer reads from it. Clients are thin.
A channel_id is the unit of conversation. The hub's WAL is keyed by channel id; expectation evaluators evaluate per channel; views project per channel.
Each AgentClient carries a default_handler that auto-acks invites and runs Agent.ask on inbound text. You can replace it with agent_client.on_envelope(callback) when you need custom logic.
The hub assigns the agent_id at registration. Use it (alice.agent_id) for routing rather than the human-readable name. The name may not be unique under a multi-tenant deployment.

Where to Next

Hub & Identity — the registry side: Hub.open, Passport, Resume, Rule, auth.
Agent Clients — the agent side: HubClient.register, default handler, custom handlers.
Channel Adapters — pick the right one: free-form, 1Q1R, round-robin, or graph-driven.