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="…")],
)

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,
)

new_rule = Rule(access=AccessBlock(outbound_to=["bob"]))
await hub.set_rule(alice.agent_id, new_rule)  # emits AUDIT_KIND_RULE_SET

from ag2.network import parse_duration

parse_duration("30s")  # 30
parse_duration("4h")   # 14400
parse_duration("2d")   # 172800

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

`Deny.error` picks the `NetworkError` subclass the hub raises (default `AccessDeniedError`) — `Deny(..., error=InboxFull)` etc. to control it. The arbiter is the **gatekeeper** (consulted *before* the state change); `HubListener` (later) is the **observer** (notified *after*). It's a different concern from `AuthAdapter` below — that authenticates *credentials* once at registration; the arbiter authorizes *actions* throughout the channel's life.

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}")

`AuthAdapter` is a `Protocol` with a `scheme` attribute and a `validate` method:

```python
class AuthAdapter(Protocol):
    scheme: str
    async def validate(self, passport: Passport, claim: dict[str, Any]) -> None: ...

from ag2.network import Expectation

Expectation(name="acks_within", on_violation="auto_close", params={"seconds": 30})

channel = await alice.open(
    type="conversation",
    target=bob.agent_id,
    knobs={
        "expectations": [
            {"name": "max_silence", "on_violation": "auto_close",
             "params": {"seconds": 600}},
        ],
    },
)

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

hub = await Hub.open(MemoryKnowledgeStore(), expectation_sweep_interval=0)

undefined

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

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,
)

undefined

The audit log is **durable when backed by `DiskKnowledgeStore`**; with `MemoryKnowledgeStore` it lives only as long as the hub.

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

undefined

Pass `context=ctx` so the task fires its events on the LLM-turn's stream — that's the stream the `TaskMirror` is attached to. Without it, the events never reach the hub.

If `capability=None` (the default), lifecycle events still go to the hub's audit log but `Resume.observed` is **not** updated. Track record is opt-in.

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")

@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

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)

@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 "..."

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,
)