Hub & Identity

The hub is the network's single source of truth: registry, audit log, channel table, write-ahead logs, expectation evaluators, sweepers. Every send, observation, and rule check goes through it.

This page covers the registry-side primitives — what you stand up before any agent connects.

Hub.open#

1
2
3
4
5
6
7
8
9

from autogen.beta.knowledge import MemoryKnowledgeStore
from autogen.beta.network import Hub

hub = await Hub.open(
    MemoryKnowledgeStore(),
    ttl_sweep_interval=30.0,        # default: 30s
    expectation_sweep_interval=10.0,# default: 10s
    invite_ack_timeout=30.0,        # default: 30s
)

Hub.open(...) is the factory; the constructor Hub(...) exists but doesn't start sweepers. Use open() unless you have a reason not to.

Identity Model#

Three small dataclasses describe an agent on the network:

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

from autogen.beta.network import Passport, Resume, ResumeExample

passport = Passport(
    name="alice",
    owner="acme",
    model="claude-sonnet-4-6",
)

resume = Resume(
    claimed_capabilities=["analysis", "policy"],
    domains=["finance"],
    summary="Senior policy analyst — scenario synthesis and rebuttal review.",
    examples=[ResumeExample(title="Q3 risk brief", note="…")],
)

Passport — stable identity#

hub.list_agents(kind=...) filters the registry by participant kind — kind="human" returns only humans, kind="agent" returns agents (and matches kind=None passports), kind=None returns everything. Non-LLM participants are registered via hc.register_human(...) — see HumanClient (HITL).

Resume — capability claims and observed track record#

Resume is the agent's network "CV" — both what it claims to do and what the hub has observed it actually doing.

Rule — per-agent governance#

Optional per-agent policy. Pass to hc.register(agent, passport, resume, rule=...).

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

from autogen.beta.network import Rule, AccessBlock, LimitsBlock, RateBlock, InboxBlock

rule = Rule(
    access=AccessBlock(
        outbound_to=["bob", "carol"],   # whitelist who this agent can address
        channel_types_allowed=["consulting", "discussion"],
    ),
    limits=LimitsBlock(
        channel_ttl_default="4h",
        delegation_depth=2,
    ),
    rate=RateBlock(
        envelopes_per_minute=60,
    ),
    inbox=InboxBlock(
        max_pending=100,
    ),
)

When a rule's check fails, the hub raises AccessDeniedError (deny by access) or InboxFull (queue cap) instead of letting the envelope through.

Authentication#

1
2
3
4
5
6
7
8
9

from autogen.beta.network import AuthAdapter, AuthRegistry, NoAuth

# default: NoAuth registered for the empty scheme
hub = await Hub.open(store)

# explicit:
auth = AuthRegistry()
auth.register("hmac", MyHMACAdapter())
hub = await Hub.open(store, auth=auth)

AuthAdapter is a Protocol:

class AuthAdapter(Protocol):
    async def verify(self, passport: Passport, credentials: AuthBlock) -> None: ...

Hub calls verify(...) at registration. Raise AuthError to reject. The default registry registers NoAuth for the empty scheme so the simple in-process flow works with no setup.

Audit Log#

Every governance-relevant event the hub processes lands in hub.audit_log (an AuditLog, which is itself a registered HubListener):

1
2
3

records = await hub.audit_log.read_all()
for r in records:
    print(r["kind"], r["at"], r)

Audit kinds (re-exported from autogen.beta.network):

The kind set is open — tenants and hub subclasses can write their own kinds via hub.audit_log.append(record). The audit log is the single trail for compliance and debugging — see Governance, Audit & Observability.

Reading Hub State#

Hub TTL & Sweepers#

Two background tasks run as long as the hub is open:

• TTL sweeper — walks active channels and expires those past created_at + ttl. Default TTL comes from Rule.limits.channel_ttl_default or the adapter's manifest.
• Expectation sweeper — walks the expectation evaluators registered for each open channel, fires ViolationHandlers on threshold breach.

Both are tunable via ttl_sweep_interval and expectation_sweep_interval. Set to 0 to disable for tests; in those cases call await hub._ttl_tick() or await hub._expectation_tick() manually for deterministic timing.

Custom sweepers#

Attach your own periodic worker — protocol-specific background work like polling a chat platform's presence list or refreshing an auth token:

1
2
3
4
5
6

async def heartbeat() -> None:
    ...  # runs every 30s

hub.register_sweeper("heartbeat", 30.0, heartbeat)
# later, e.g. on a config reload:
await hub.unregister_sweeper("heartbeat")

• register_sweeper(name, interval_seconds, fn) is synchronous (it only updates bookkeeping). If the hub has already started (Hub.open(...) calls start() for you), the sweeper begins immediately; if you registered it before start(), it begins then.
• unregister_sweeper(name) is async — it awaits clean cancellation of the running task. Unknown names are a no-op.
• A duplicate name raises ValueError (call unregister_sweeper first to replace); a non-positive interval raises ValueError.
• hub.close() stops every custom sweeper automatically — you don't need to track them yourself.

Closing Down#

await hub.close()

Cancels all sweeper tasks, closes the underlying store, drains pending I/O. Always pair Hub.open(...) with hub.close() (typically in a try/finally).