# Human Clients (HITL)
A `HumanClient` is a **non-LLM participant** on the network — the human-in-the-loop primitive. It is a client in the network, just like LLM agents, so the hub routes envelopes to it exactly the same way.
Your application supplies the UI; the framework supplies the participant.
Use it whenever a person (one or many) needs to join a channel — answering a `consulting` request, taking a turn in a `discussion`, seeding a `workflow`, or just chatting in a `conversation`.
## Registering
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
``` | ```
from autogen.beta.network import HubClient, LocalLink, Passport
hc = HubClient(LocalLink(hub), hub=hub)
human = await hc.register_human(
Passport(name="operator"),
resume=None, # optional — Resume() defaults
rule=None, # optional — Rule(...) for governance, same as agents
auto_ack_invites=True, # auto-accept channel invites (see below)
)
``` |
`register_human` runs the same UUID-stamping and persistence path as `hc.register(...)`, then forces `passport.kind = "human"` so the participant is discoverable as a human:
```
await hub.list_agents(kind="human") # -> [Passport(name="operator", kind="human", ...)]
await hub.list_agents(kind="agent") # agents only (also matches kind=None)
```
`hc.register(...)` rejects `Passport(kind="human")`, ensure you use `register_human` for `HumanClient`s.
## Receiving — push or pull
A `HumanClient` exposes inbound envelopes two ways. Both see every inbound envelope; use whichever fits your UI (or both at once).
### Push — `on_envelope`
Register a coroutine; it fires once per inbound envelope. Multiple callbacks compose in registration order. A callback that raises exceptions is logged and **never** propagates to the hub's dispatch path — a buggy UI cannot break the network.
| | |
| --- | --- |
| ```
1
2
3
4
5
``` | ```
async def on_inbound(envelope) -> None:
await ui.push_event(envelope) # forward to a websocket, queue, etc.
human.on_envelope(on_inbound)
human.remove_envelope_callback(on_inbound) # detach later
``` |
### Pull — `next_envelope` / `envelopes`
Block until the next matching envelope arrives, or iterate the inbound stream:
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
``` | ```
from autogen.beta.network import EV_TEXT
# Wait for the next text reply from a specific peer.
reply = await human.next_envelope(
predicate=lambda e: e.event_type == EV_TEXT and e.sender_id == peer_id,
timeout=60.0, # raises asyncio.TimeoutError if exceeded
)
# Or stream everything until disconnect.
async for envelope in human.envelopes():
...
# Channel-scoped wait (symmetric with AgentClient.wait_for_channel_event):
env = await human.wait_for_channel_event(
channel_id=channel.channel_id,
predicate=lambda e: e.event_type == EV_TEXT,
timeout=300.0,
)
``` |
Envelopes that don't match a `next_envelope` predicate are discarded — use `on_envelope` if you want to observe everything _and_ await something specific.
## Sending
Outbound mirrors `AgentClient`. `human.open(...)` returns the same `Channel` handle, so multi-turn channel code is identical whether the initiator is a human or an agent.
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
11
``` | ```
from autogen.beta.network import CONVERSATION_TYPE
channel = await human.open(type=CONVERSATION_TYPE, target=expert.agent_id)
await channel.send("Hi — what's a good first ML concept to learn?")
await channel.close(reason="done")
# Convenience for an existing channel id:
await human.send(channel_id, "another message")
# Escape hatch for adapter-shaped envelopes (e.g. a workflow EV_PACKET seed):
await human.post_envelope(envelope)
``` |
| Call | Notes |
| --- | --- |
| `await human.open(type=, target=, ttl=, knobs=, intent=, labels=)` | Open a channel as the initiator → `Channel`. `target` accepts peer names or agent ids. |
| `await human.send(channel_id, text, audience=, causation_id=)` | Post an `EV_TEXT` envelope. |
| `await human.post_envelope(envelope)` | Post an arbitrary envelope (stamps `sender_id` if blank). |
| `await human.close_channel(channel_id, reason=)` | Close a channel this human is in. |
| `await human.disconnect()` | Stop accepting deliveries; wakes any blocked `next_envelope` / `envelopes` consumers. Idempotent — call it in your shutdown path. |
## Channel invites — `auto_ack_invites`
When an agent opens a channel to a human, the hub waits for the human's `EV_CHANNEL_INVITE_ACK` before the channel reaches `ACTIVE`. With `auto_ack_invites=True` (the default) the `HumanClient` acks automatically the moment the invite arrives — the channel handshake completes with no UI round-trip, exactly like the default agent handler.
Pass `auto_ack_invites=False` if you want a human to _decide_ whether to join (an "accept invite?" prompt). If so, your UI is responsible for emitting the ack:
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
``` | ```
from autogen.beta.network import EV_CHANNEL_INVITE, EV_CHANNEL_INVITE_ACK, Envelope
human = await hc.register_human(Passport(name="operator"), auto_ack_invites=False)
async def gate_invites(envelope) -> None:
if envelope.event_type != EV_CHANNEL_INVITE:
return
if await ui.confirm(f"Join channel {envelope.channel_id}?"):
await human.post_envelope(Envelope(
channel_id=envelope.channel_id,
sender_id=human.agent_id,
event_type=EV_CHANNEL_INVITE_ACK,
event_data={"channel_id": envelope.channel_id},
causation_id=envelope.envelope_id,
))
# otherwise let the hub's invite-ack timeout close the channel
human.on_envelope(gate_invites)
``` |
## Hooking up a UI
The framework deliberately doesn't pick an input modality — you bridge the `HumanClient` to whatever UI you have. Two common shapes:
### A web app / websocket bridge (push out, RPC in)
Forward inbound envelopes to the client over a websocket; turn UI messages into sends.
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
``` | ```
async def serve(websocket, human):
# outbound: hub -> browser
async def to_browser(envelope) -> None:
await websocket.send_json({
"channel": envelope.channel_id,
"from": envelope.sender_id,
"type": envelope.event_type,
"data": envelope.event_data,
})
human.on_envelope(to_browser)
# inbound: browser -> hub
try:
async for msg in websocket:
payload = msg.json()
if payload["action"] == "send":
await human.send(payload["channel"], payload["text"])
elif payload["action"] == "open":
await human.open(type=payload["type"], target=payload["target"])
elif payload["action"] == "close":
await human.close_channel(payload["channel"], reason="user_closed")
finally:
human.remove_envelope_callback(to_browser)
await human.disconnect()
``` |
### A console / CLI loop (pull)
`input()` is blocking — run it off the event loop with `asyncio.to_thread` so the network stays responsive while the user types.
| | |
| --- | --- |
| ```
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
``` | ```
import asyncio
from autogen.beta.network import CONVERSATION_TYPE, EV_TEXT
channel = await human.open(type=CONVERSATION_TYPE, target=expert.agent_id)
while True:
text = (await asyncio.to_thread(input, "you> ")).strip()
if not text or text.lower() in {"quit", "exit"}:
break
await channel.send(text)
try:
reply = await human.next_envelope(
predicate=lambda e: e.event_type == EV_TEXT and e.sender_id == expert.agent_id,
timeout=60.0,
)
print(f"expert> {reply.event_data['text']}")
except asyncio.TimeoutError:
print("expert> (no reply within 60s)")
await channel.close(reason="operator_done")
await human.disconnect()
``` |
Drain rate is yours to manage
The pull queue is unbounded by design — the embedder controls how fast it's drained via the UI. If it grows pathologically, the application has a UI bug to fix. Always call `human.disconnect()` on shutdown so blocked consumers wake up instead of hanging.