AG2 Network — Page 8 (BYO Adapter)

Network canvas · the protocol + a live example pip (init.) r1 r2 r3

manifest

class-level

ChannelManifest

idle

→

initial_state

once / channel

(metadata) → AdapterState

idle

→

validate_send

per envelope

(meta, env, state) → None

idle

→

fold

per envelope

(env, state) → AdapterState

idle

→

on_accepted

per envelope

(meta, env, state) → AdapterResult

idle

pip

initiator

respondent

No channel yet. Once pip.open(type="poll", target=[r1,r2,r3]) runs, messages appear here.

PollState

question_sent False

replied { }

channel no channel

BYO Adapter walkthrough 0 of 5

step 0

Five hooks form the ChannelAdapter protocol. Every adapter — including the four AG2 ships — implements these. Today we'll write a new one: PollAdapter. The hooks above are idle; the example channel hasn't opened yet.

Protocol

0 meet the 5 hooks · no channel yet

1 ① ChannelManifest · the contract

2 ② initial_state() → PollState()

3 ③ validate_send · pip posts question

4 ④ fold · replies grow the state

5 ⑤ on_accepted · auto_close on last

PollAdapter · the full class ~50 lines · text-only I/O uses default helpers

from dataclasses import dataclass, field
from autogen.beta.network import (
    ChannelAdapter, ChannelMetadata, ChannelManifest,
    ParticipantSchema, Expectation,
    Envelope, AdapterResult, EV_TEXT,
)
from autogen.beta.network.adapters.base import (
    default_extract_turn_input,
    default_build_round_envelope,
    default_render_envelope,
)


@dataclass(slots=True)
class PollState:
    question_sent: bool = False
    replied: frozenset[str] = field(default_factory=frozenset)


class PollAdapter:
    manifest = ChannelManifest(
        type="poll",
        version=1,
        participants=ParticipantSchema(
            min=2, max=10,
            roles=["initiator", "respondent"],
        ),
        knobs_schema={},
        default_view_policy="full_transcript",
        expectations=[
            Expectation(
                name="reply_within",
                on_violation="auto_close",
                params={"seconds": 600},
            ),
        ],
    )

    def initial_state(self, metadata) -> PollState:
        return PollState()

    def validate_create(self, metadata) -> None:
        if len(metadata.participants) < 2:
            raise ValueError("poll requires >= 2 participants")

    def validate_send(self, metadata, envelope, state: PollState) -> None:
        initiator = metadata.creator_id
        if not state.question_sent:
            if envelope.from_ != initiator:
                raise PermissionError("only the initiator can post")
        else:
            if envelope.from_ == initiator:
                raise PermissionError("initiator already posted")
            if envelope.from_ in state.replied:
                raise PermissionError("already replied")

    def fold(self, envelope, state: PollState) -> PollState:
        if envelope.event_type != EV_TEXT:
            return state
        if not state.question_sent:
            return PollState(question_sent=True, replied=state.replied)
        return PollState(
            question_sent=True,
            replied=state.replied | {envelope.from_},
        )

    def on_accepted(self, metadata, envelope, state: PollState) -> AdapterResult:
        respondents = set(metadata.participants) - {metadata.creator_id}
        if state.question_sent and state.replied >= respondents:
            return AdapterResult(
                next_state=None,
                auto_close_reason="poll_complete",
            )
        return AdapterResult(next_state=None, auto_close_reason=None)

    # text-only I/O — reuse the framework defaults
    extract_turn_input    = default_extract_turn_input
    build_round_envelope  = default_build_round_envelope
    render_envelope       = default_render_envelope

    def default_view_policy(self, metadata, participant_id):
        from autogen.beta.network import FullTranscript
        return FullTranscript()

Register & use · hub.register_adapter(PollAdapter()) that's the entire extension point — no factories, no plugin manifests

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

hub = await Hub.open(MemoryKnowledgeStore())
hub.register_adapter(PollAdapter())            # keyed by (manifest.type, manifest.version)

# now any agent can open type="poll" channels:
channel = await pip.open(
    type="poll", version=1,
    target=[r1.agent_id, r2.agent_id, r3.agent_id],
)
await channel.send("What's your one-word focus for next quarter?")
# r1, r2, r3 each reply once → channel auto-closes with reason="poll_complete"

That's the whole protocol. Five methods (four really, since text-only I/O reuses framework defaults). A pure state machine. No inheritance, no factory, no plugin manifest — just instantiate your class and call hub.register_adapter(...). Re-registering doesn't affect in-flight channels, because each channel snapshots its manifest at creation. Build whatever turn-taking model your domain actually needs.