AG2 Network — Governance
Everything hub-side: identity, per-agent rules, expectations, audit, and task observation. The hub is the single source of truth — every send goes through it, every observation reads from it, every policy is checked there.
Prerequisite: read
first. This skill assumes you know
,
,
, the channel lifecycle, and the
agent_client.register(...)
flow.
When to use
Load this skill when the user needs to:
- Limit who can talk to whom ( / )
- Declare a token-bucket rate intent ( / ) — stored on the rule but not enforced by the in-process hub
- Cap inbox size to prevent flooding ( / ) — and get an early-warning signal before the cap ( / )
- Set channel TTL defaults, concurrency caps, or delegation depth ( / )
- Plug in custom access / routing logic — JWT scopes, per-tenant quotas, federation ( / )
- Authenticate agents at registration ()
- Tune the channel-close timing (, , , )
- Read or query the audit log for compliance — or stream live state changes to metrics / alerting ()
- Add a custom periodic task to the hub ()
- Build a capability track record on each agent ()
- Route based on which agents have demonstrably done a task (e.g. "send to whichever researcher has the lowest ")
Identity — what every agent carries
Three dataclasses describe an agent on the network. The tenant supplies most fields; the hub stamps the rest.
python
from ag2.network import Passport, Resume, ResumeExample
passport = Passport(
name="alice", # required, unique within the hub
owner="acme", # optional, tenant id for multi-tenant deployments
model="claude-sonnet-4-6", # optional, surfaces on peer-lookup results
)
resume = Resume(
claimed_capabilities=["analysis", "policy"],
domains=["finance"],
summary="Senior policy analyst — scenario synthesis and rebuttal review.",
examples=[ResumeExample(title="Q3 risk brief", note="…")],
)
| Field | Source |
|---|
| tenant (required, unique) |
| hub-stamped at registration; use for routing |
| hub-stamped ISO-Z timestamp |
Resume.claimed_capabilities
| tenant (free-form strings: , , …) |
| tenant — indexed for peer lookup |
| hub-mutated per-capability (n / completed / failed / expired / p50_latency_ms) |
| hub-stamped ISO-Z, refreshed on mutation |
The
field is the agent's track record. It grows automatically as the agent runs capability-tagged tasks (see "Task observation" below).
Per-agent rules
Pass a
at registration to govern an agent's behaviour on the network:
python
from ag2.network import (
Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock, ChannelTypeAccess,
)
rule = Rule(
access=AccessBlock(
outbound_to=["bob", "carol"], # whitelist of recipients (names or ids)
channel_types=ChannelTypeAccess(
initiate=["consulting", "discussion"],
accept=["consulting", "discussion"],
),
),
limits=LimitsBlock( # rate + inbox nest INSIDE limits
channel_ttl_default="4h", # default TTL for channels this agent creates
delegation_depth=2, # max recursion through sub-task delegation
rate=RateBlock(per_minute=60, burst=10),
inbox=InboxBlock(max_pending=100), # cap inbound queue depth
),
)
alice = await alice_hc.register(
Agent("alice", config=config),
Passport(name="alice"),
Resume(),
rule=rule,
)
A
has
two top-level blocks:
(
) and
(
).
and
nest
inside (as
and
).
| Block | Lives at | Controls | Failure mode |
|---|
| (top-level) | Who this agent can address () / accept from (); channel types it can create/join | |
| (top-level) | TTL defaults; ; / | |
| (nested) | Token-bucket values (, ) | Stored but not enforced in-process ( disables by default) |
| (nested) | Inbound queue depth () | to the sender |
When a rule check fails the hub raises the matching error from
or
; the envelope never lands on the WAL. Rule
changes are audited (kind
via
), but a send
denial is
not written to the built-in audit log — it surfaces via the
listener fan-out (the built-in
implements no
handler). Register a
if you need to capture rejections. The component that
runs those checks — and the seam where you'd plug in something other than rule data — is the
arbiter, below.
Updating a rule after registration
python
new_rule = Rule(access=AccessBlock(outbound_to=["bob"]))
await hub.set_rule(alice.agent_id, new_rule) # emits AUDIT_KIND_RULE_SET
Parsing duration strings
LimitsBlock.channel_ttl_default
accepts a string parsed by
:
python
from ag2.network import parse_duration
parse_duration("30s") # 30
parse_duration("4h") # 14400
parse_duration("2d") # 172800
,
,
,
suffixes; a bare integer (or
) is treated as seconds; empty string returns
. Returns an
number of seconds.
The arbiter — swappable access & routing
The hub doesn't enforce
s with inline
checks anymore; it delegates every access / routing decision to a
— a Protocol with one method per decision point, each returning
or
Deny(reason, error=<NetworkError subclass>)
:
| Method | Consulted before… | Default error |
|---|
authorize_register(passport, resume, rule)
| committing a registration | |
authorize_channel_open(manifest, creator, creator_rule, invitees, invitee_rules, active_creator_channels)
| creating a channel (invitee + creator ) | |
authorize_send(envelope, sender, sender_rule, recipients)
| appending an envelope to the WAL (outbound access + delegation depth) | |
authorize_inbox(envelope, recipient, recipient_rule, current_pending)
| enqueuing into a recipient's inbox (capacity) | |
authorize_dispatch(envelope, sender, recipient, recipient_rule)
| dispatching one delivery | |
resolve_unknown_audience(envelope, unknown_ids)
| dispatching to ids the hub doesn't know — returns (drop silently — the single-hub default) or a replacement id list (federation hook) | — |
The default is
— it enforces the per-agent
:
(outbound/inbound name globs, channel types) plus the
caps it actually checks (
,
,
).
is
not enforced. This is exactly what the hub did inline before this seam existed. If you only use
, you never touch the arbiter.
Swap it to layer your own logic — JWT scopes, per-tenant quotas, federation routing — on top of (or instead of) the rule data.
returns
for everything, so a subclass that overrides one gate would
allow the rest — to keep rule enforcement, delegate to a
instance:
python
from ag2.network import HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny
class ScopedArbiter(BaseHubArbiter):
def __init__(self, inner: HubArbiter) -> None:
self._inner = inner # the rule checks
async def authorize_send(self, envelope, sender, sender_rule, recipients):
if not _token_has_scope(sender, "net.send"):
return Deny("missing net.send scope") # → AccessDeniedError back to the caller
return await self._inner.authorize_send(envelope, sender, sender_rule, recipients)
# authorize_register / _channel_open / _inbox / _dispatch / resolve_unknown_audience
# all fall through to BaseHubArbiter's Allow() — explicitly re-delegate any you want enforced.
hub.register_arbiter(ScopedArbiter(RuleBasedArbiter())) # one active arbiter; replaces the prior one
# hub.arbiter → the active instance (read-only; handy in tests)
picks the
subclass the hub raises (default
) —
Deny(..., error=InboxFull)
etc. to control it. The arbiter is the
gatekeeper (consulted
before the state change);
(later) is the
observer (notified
after). It's a different concern from
below — that authenticates
credentials once at registration; the arbiter authorizes
actions throughout the channel's life.
Authentication
By default the hub uses
— a
-only registry (scheme
) that accepts every claim, so every registration succeeds without credentials. The scheme is selected per-passport via
(an
field, defaulting to
); the credentials live in
(a
). For production:
python
from ag2.network import AuthAdapter, AuthRegistry, NoAuth, ApiKeyAuth, AuthError, Hub
from ag2.network import AuthBlock, Passport
from ag2.knowledge import MemoryKnowledgeStore
import hmac
from typing import Any
class HMACAuth:
scheme = "hmac" # class-level scheme label
async def validate(self, passport: Passport, claim: dict[str, Any]) -> None:
expected = self._sign(passport.name)
token = claim.get("token", "")
if not hmac.compare_digest(expected, token):
raise AuthError(f"bad hmac for {passport.name}")
# AuthRegistry takes a LIST of adapters at construction (keyed by .scheme).
registry = AuthRegistry([NoAuth(), HMACAuth()])
hub = await Hub.open(MemoryKnowledgeStore(), auth=registry)
is a
with a
attribute and a
method:
python
class AuthAdapter(Protocol):
scheme: str
async def validate(self, passport: Passport, claim: dict[str, Any]) -> None: ...
Raise
to reject. At registration the hub looks up the adapter by
, calls
adapter.validate(passport, passport.auth.claim)
, and records
AUDIT_KIND_AGENT_REGISTERED
on success. Remote-agent passports skip the local auth check. The library ships
(accept-all, the default) and
ApiKeyAuth(keys=..., resolver=...)
(constant-time token compare against
).
Expectations — channel-level SLAs
Every adapter ships defaults in its manifest. The expectation sweeper task evaluates them every
expectation_sweep_interval
(default 10s) and dispatches violations to handlers.
Built-in evaluators
ships exactly
three evaluators:
| Name | Class | Default | Threshold |
|---|
| | 30 | All still-pending invitees must ack within of channel creation (only while the channel is ). |
| | 600 | A participant addressed by an must reply within (only while ). |
| | 3600 | The channel has no content envelope from anyone for (channel-wide). |
The
and
adapters declare
expectations on their manifests, but
no built-in evaluator ships — and
/
are
not built-in handlers. The sweeper silently skips any expectation whose
has no registered evaluator or whose
has no registered handler (
does
and
s on
). To make
/
/
active, register your own evaluator (
register_expectation_evaluator
) and handler (
register_violation_handler
).
Default expectations per adapter
| Adapter | Defaults |
|---|
| acks_within(30s, auto_close)
reply_within(600s, auto_close)
|
| max_silence(3600s, audit)
|
| , |
| , turn_within(600s, auto_close)
|
Violation handlers
python
from ag2.network import Expectation
Expectation(name="acks_within", on_violation="auto_close", params={"seconds": 30})
ships exactly
three handlers, keyed by
:
| Handler class | Effect |
|---|
| | No-op handler. The actual audit record (AUDIT_KIND_EXPECTATION_VIOLATED
|
| | Post to every channel participant. Channel continues. (Audit is still written by the listener.) |
| | Close the channel with reason="expectation_violated:<name>"
|
There is
no built-in or handler — those names appear only on the
/
manifests and are no-ops until you register a handler for them.
Overriding adapter defaults
Pass
in the channel knobs to replace the adapter's defaults:
python
channel = await alice.open(
type="conversation",
target=bob.agent_id,
knobs={
"expectations": [
{"name": "max_silence", "on_violation": "auto_close",
"params": {"seconds": 600}},
],
},
)
Custom evaluators
python
from ag2.network import EV_TEXT, Expectation
from ag2.network import ExpectationContext, Violation
class TooManyMessagesEvaluator:
name = "too_many_messages"
# Signature: evaluate(self, expectation, context) -> Violation | None
def evaluate(self, expectation: Expectation, context: ExpectationContext) -> Violation | None:
threshold = int(expectation.params["max"])
text_count = sum(1 for e in context.wal if e.event_type == EV_TEXT)
if text_count > threshold:
return Violation(
expectation=expectation, # the Expectation object, not a string
violator_ids=[], # channel-wide
detail={"text_count": text_count, "threshold": threshold},
)
return None
is
Violation(expectation: Expectation, violator_ids: list[str] = [], detail: dict = {})
—
is the
object (it carries
,
,
);
is
not a field (the hub already knows it). Evaluators are pure functions over channel state — no I/O, no mutation — so they're trivially testable. Register via
hub.register_expectation_evaluator(TooManyMessagesEvaluator())
.
Deterministic testing
python
hub = await Hub.open(MemoryKnowledgeStore(), expectation_sweep_interval=0)
# Manually advance state and tick:
clock.advance(45)
await hub._expectation_tick() # operator API (leading underscore by convention)
Audit log
The hub maintains an append-only audit log (
instance), exposed via the public
property (the internal attribute is
; swap the instance with
hub.replace_audit_log(...)
):
python
records = await hub.audit_log.read_all()
for r in records:
print(r["kind"], r["at"], r)
Each record is a plain dict with at minimum
and
(ISO-Z timestamp); kind-specific fields appear alongside.
Audit kinds
python
from ag2.network import (
AUDIT_KIND_AGENT_REGISTERED,
AUDIT_KIND_AGENT_UNREGISTERED,
AUDIT_KIND_RESUME_SET,
AUDIT_KIND_SKILL_SET,
AUDIT_KIND_RULE_SET,
AUDIT_KIND_CHANNEL_CREATED,
AUDIT_KIND_CHANNEL_CLOSED,
AUDIT_KIND_CHANNEL_EXPIRED,
AUDIT_KIND_TASK_TERMINATED,
AUDIT_KIND_EXPECTATION_VIOLATED,
)
| Kind | When | Common fields |
|---|
AUDIT_KIND_AGENT_REGISTERED
| | , , |
AUDIT_KIND_AGENT_UNREGISTERED
| | |
| | Source: or |
| | Updated skill markdown |
| | The new |
AUDIT_KIND_CHANNEL_CREATED
| | , manifest type/version, participants |
AUDIT_KIND_CHANNEL_CLOSED
| Any close route | |
AUDIT_KIND_CHANNEL_EXPIRED
| TTL sweeper | TTL details |
AUDIT_KIND_TASK_TERMINATED
| reached terminal state via | , , , |
AUDIT_KIND_EXPECTATION_VIOLATED
| Expectation evaluator's threshold elapsed | , , evaluator detail |
Common queries
python
# All violations on the system.
violations = [r for r in await hub.audit_log.read_all()
if r["kind"] == AUDIT_KIND_EXPECTATION_VIOLATED]
# Everything that happened on one channel.
channel_records = [r for r in await hub.audit_log.read_all()
if r.get("channel_id") == channel_id]
# All registrations for one tenant.
acme_agents = [r for r in await hub.audit_log.read_all()
if r["kind"] == AUDIT_KIND_AGENT_REGISTERED
and r.get("owner") == "acme"]
The audit log is
durable when backed by ; with
it lives only as long as the hub.
Hub listeners — live programmatic observability
The audit log is the
durable record. For
live reactions to hub state changes — push to a metrics backend, stream to a dashboard, alert an on-call — register a
: a read-only Protocol the hub fans out to after every state transition has committed. (The built-in audit log is itself one of these listeners.)
| Method (exact signature) | Fires when |
|---|
on_envelope_posted(envelope, metadata)
| an envelope was accepted, WAL-appended, folded, and dispatched |
on_envelope_rejected(envelope, reason)
| the arbiter / validation denied a send ( is the typed ) |
on_dispatch_failed(envelope, recipient_id, reason)
| delivery to one recipient raised ( is a ) |
on_channel_event(channel_id, kind, payload)
| ∈ / / / / |
on_agent_event(agent_id, kind, payload)
| ∈ / / / / / |
on_expectation_fired(channel_id, expectation, violation)
| an expectation evaluator emitted a |
on_turn_failed(channel_id, agent_id, envelope_id, exc)
| an agent's notify-handler turn raised (the default handler routes failures here) |
on_task_event(task_id, kind, payload)
| a lifecycle event was observed ( ∈ / / / / / / ) |
on_inbox_pressure(agent_id, pending, cap)
| a recipient's inbox first crosses (fires once per crossing, not per envelope) |
All methods are
; the hub awaits them sequentially in registration order, each wrapped in
— a buggy listener can't stall dispatch. Keep them fast (queue I/O onto your own task). Subclass
(every method is a
) and override only what you need:
python
from ag2.network import BaseHubListener
class MetricsListener(BaseHubListener):
async def on_envelope_posted(self, envelope, metadata):
statsd.incr(f"net.envelope.{envelope.event_type}")
async def on_inbox_pressure(self, agent_id, pending, cap):
statsd.gauge(f"net.inbox.{agent_id}", pending / cap)
async def on_turn_failed(self, channel_id, agent_id, envelope_id, exc):
sentry.capture_exception(exc)
hub.register_listener(MetricsListener()) # hub.unregister_listener(inst) to detach
Two related hub-subclass seams:
- hooks on itself — the same method set exists as empty methods on ; a subclass can override them directly (the fan-out invokes the bound method alongside registered listeners). Use a subclass when the observability is the hub variant you're shipping; use for pluggable add-ons.
hub.register_sweeper(name, interval_seconds, fn)
is governed by
(an
field) — an absolute pending-count threshold (
).
(the default) auto-resolves to
int(limits.inbox.max_pending * 0.8)
;
disables the signal. It's the early-warning sibling of the hard
(the cap is
, enforced by the arbiter's
).
Task observation — building the track record
Capability-tagged tasks update an agent's
Resume.observed[capability]
automatically. This is how the network knows that "bob has completed 47 research tasks at a 4.2s median latency."
Tagging a task
agent.task(..., capability="X")
accepts a free-form capability string:
python
# `.tool` and `.task(...)` live on `Agent`, not on the `AgentClient` returned
# by `hc.register(...)`. So decorate the Agent before registering.
@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
async with worker_agent.task(
f"research: {topic}",
capability="research",
context=ctx,
) as task:
await task.progress({"step": "gather"})
# ... do work ...
await task.complete({"items_found": 7})
return "researched"
worker = await worker_hc.register(worker_agent, Passport(name="worker"), Resume())
Pass
so the task fires its events on the LLM-turn's stream — that's the stream the
is attached to. Without it, the events never reach the hub.
If
(the default), lifecycle events still go to the hub's audit log but
is
not updated. Track record is opt-in.
Reading the track record
python
resume = await hub.get_resume(bob.agent_id)
stat = resume.observed.get("research")
if stat:
print(f"completed={stat.completed}/{stat.n} "
f"failed={stat.failed} "
f"p50_latency={stat.p50_latency_ms}ms")
python
@dataclass(slots=True)
class ObservedStat:
n: int = 0 # total terminal events
completed: int = 0
failed: int = 0
expired: int = 0
p50_latency_ms: int | None = None # rolling median of started_at → completed_at
Latency is computed from
to the terminal event time, using the hub's clock. With a
in tests you can construct deterministic latencies.
Where fits
The default handler auto-attaches a
per turn, scoped to the active channel. The mirror subscribes to
/
/
/
/
events on the LLM-turn's stream, forwards each as an
envelope to the hub, and on terminal events with a
tag calls
Hub.record_observation(...)
to update
.
You only attach
manually if you've written a custom handler:
python
from ag2.network import TaskMirror
from ag2.stream import MemoryStream
mirror = TaskMirror(
hub_client=client._hub_client,
owner_id=client.agent_id,
channel_id=metadata.channel_id,
)
stream = MemoryStream()
sub_ids = mirror.attach(stream)
try:
await client.agent.ask(text, stream=stream)
finally:
mirror.detach(stream, sub_ids)
The mirror is attached per turn, not per agent — a new one for each inbound envelope. It also swallows hub-forwarding errors silently; a flaky hub connection should not crash the LLM turn.
When to skip the capability tag
Tag only when:
- The task represents a capability you want to track in the agent's resume.
- Failure / latency signals are operationally meaningful (driving routing, alerting, peer ranking).
Untagged tasks still get full lifecycle audit records — just no
update. Use untagged tasks for internal book-keeping that doesn't represent an externally-visible capability.
Cross-cutting pattern: multi-capability worker
python
@worker_agent.tool
async def research(topic: str, ctx: Context) -> str:
async with worker_agent.task(f"research: {topic}", capability="research", context=ctx) as t:
# ...
return "..."
@worker_agent.tool
async def summarise(text: str, ctx: Context) -> str:
async with worker_agent.task("summarise", capability="summarisation", context=ctx) as t:
# ...
return "..."
After a few channels,
holds both
and
s, each tracked independently. A peer-discovery query (
peers(action="find", capability="research")
— see
ag2-network-tools-and-views
) can then rank by latency or completion rate.
Reading hub state
| Call | Returns |
|---|
await hub.get_channel(channel_id)
| snapshot (state, participants, close_reason) |
await hub.get_resume(agent_id)
| Current (including ) |
await hub.get_passport(agent_id)
| Current |
await hub.list_agents(kind=None)
| Registered passports; / / filters by |
await hub.read_wal(channel_id)
| Ordered list of s in that channel |
await hub.audit_log.read_all()
| Every audit record |
| The active (read-only) |
The hub stamps
on every mutation, so you can detect stale views by comparing timestamps. For
push (vs. these
pull calls), register a
.
Quick reference — imports
python
from ag2.network import (
# Identity
Passport, Resume, ResumeExample, ObservedStat,
# Rules
Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock,
ChannelTypeAccess, parse_duration,
# Arbiter (swappable access / routing seam)
HubArbiter, BaseHubArbiter, RuleBasedArbiter, Allow, Deny,
# Listeners (live observability) — hub.register_listener(...)
HubListener, BaseHubListener,
# Auth
AuthAdapter, AuthRegistry, AuthBlock, NoAuth, ApiKeyAuth,
# Expectations
Expectation,
ExpectationEvaluator, ExpectationContext,
AcksWithinEvaluator, ReplyWithinEvaluator, MaxSilenceEvaluator,
AuditHandler, NotifyChannelHandler, AutoCloseHandler,
Violation, ViolationHandler,
default_evaluators, default_handlers,
# Audit kinds
AUDIT_KIND_AGENT_REGISTERED,
AUDIT_KIND_AGENT_UNREGISTERED,
AUDIT_KIND_RESUME_SET,
AUDIT_KIND_SKILL_SET,
AUDIT_KIND_RULE_SET,
AUDIT_KIND_CHANNEL_CREATED,
AUDIT_KIND_CHANNEL_CLOSED,
AUDIT_KIND_CHANNEL_EXPIRED,
AUDIT_KIND_TASK_TERMINATED,
AUDIT_KIND_EXPECTATION_VIOLATED,
RESUME_SOURCE_OBSERVED, RESUME_SOURCE_TENANT,
# Task observation
TaskMirror,
# Errors — the full family (no `LimitsExceeded`; limit/access denials raise AccessDeniedError, inbox-full raises InboxFull)
NetworkError, AccessDeniedError, AuthError, InboxFull, NotFoundError, ProtocolError,
)