Skip to content

Session

autogen.beta.network.client.session.Session #

Session(*, metadata, client)

Per-participant handle for a session.

Constructed by AgentClient.open(...) / hydrated when an EV_SESSION_OPENED lands. The metadata attribute is a snapshot — call :meth:info to refresh from the hub.

Source code in autogen/beta/network/client/session.py 
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    *,
    metadata: SessionMetadata,
    client: "AgentClient",
) -> None:
    # __init__ stores params; no side effects.
    self._metadata = metadata
    self._client = client

session_id property #

session_id

metadata property #

metadata

state property #

state

send async #

send(content, *, audience=None, causation_id=None, event_type=EV_TEXT, event_data=None, depth=None)

Post an envelope into this session.

audience=None broadcasts within the session (all participants except sender). content is the substantive body for EV_TEXT envelopes; for non-text events pass event_data and event_type instead. depth overrides the default 0 so callers (e.g. delegate) can stamp the delegation hop count for Rule.limits.delegation_depth enforcement.

Source code in autogen/beta/network/client/session.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def send(
    self,
    content: str,
    *,
    audience: list[str] | None = None,
    causation_id: str | None = None,
    event_type: str = EV_TEXT,
    event_data: dict[str, Any] | None = None,
    depth: int | None = None,
) -> str:
    """Post an envelope into this session.

    ``audience=None`` broadcasts within the session (all
    participants except sender). ``content`` is the substantive
    body for ``EV_TEXT`` envelopes; for non-text events pass
    ``event_data`` and ``event_type`` instead. ``depth`` overrides
    the default 0 so callers (e.g. ``delegate``) can stamp the
    delegation hop count for ``Rule.limits.delegation_depth``
    enforcement.
    """
    if event_data is None:
        event_data = {"text": content}
    envelope = Envelope(
        session_id=self.session_id,
        sender_id=self._client.agent_id,
        audience=audience,
        event_type=event_type,
        event_data=event_data,
        causation_id=causation_id,
        depth=depth if depth is not None else 0,
    )
    return await self._client.send_envelope(envelope)

info async #

info()

Re-fetch metadata from the hub (refreshes cached state).

Source code in autogen/beta/network/client/session.py 
86
87
88
89
90
async def info(self) -> SessionMetadata:
    """Re-fetch metadata from the hub (refreshes cached state)."""
    refreshed = await self._client._hub_client.get_session(self.session_id)
    self._metadata = refreshed
    return refreshed

close async #

close(reason='')

Close the session. Auto-cascades expiry to non-terminal tasks.

Source code in autogen/beta/network/client/session.py 
92
93
94
async def close(self, reason: str = "") -> SessionMetadata:
    """Close the session. Auto-cascades expiry to non-terminal tasks."""
    return await self._client._hub_client.close_session(self.session_id, reason=reason)

is_terminal #

is_terminal()
Source code in autogen/beta/network/client/session.py 
96
97
def is_terminal(self) -> bool:
    return self._metadata.is_terminal()