SessionMetadata(session_id, manifest, creator_id, participants, state, created_at, expires_at=None, closed_at=None, close_reason='', parent_session_id=None, knobs=dict(), labels=dict(), required_acks=None, pending_acks=list(), rejected_by=list())
Mutable lifecycle record for a session.
manifest is the snapshot taken at create time — re-registering an adapter at a new version does not retroactively change this. knobs are adapter-specific (e.g. {"ordering": "round_robin"}); labels is the catch-all for tenant annotations and includes "intent" if the creator passed one.
pending_acks and rejected_by are populated at session creation and frozen once the session transitions to ACTIVE (quorum reached) or fails creation. The handshake is all-or-nothing for both 2-party (consulting) and multi-party (discussion, conversation, workflow) sessions: any reject fails creation.
Source code in autogen/beta/network/session.py
| def to_dict(self) -> dict[str, Any]:
return {
"session_id": self.session_id,
"manifest": self.manifest.to_dict(),
"creator_id": self.creator_id,
"participants": [p.to_dict() for p in self.participants],
"state": self.state.value,
"created_at": self.created_at,
"expires_at": self.expires_at,
"closed_at": self.closed_at,
"close_reason": self.close_reason,
"parent_session_id": self.parent_session_id,
"knobs": dict(self.knobs),
"labels": dict(self.labels),
"required_acks": self.required_acks,
"pending_acks": list(self.pending_acks),
"rejected_by": list(self.rejected_by),
}
|
Source code in autogen/beta/network/session.py
| @classmethod
def from_dict(cls, data: dict[str, Any]) -> "SessionMetadata":
payload = dict(data)
manifest = payload.get("manifest")
if isinstance(manifest, dict):
payload["manifest"] = SessionManifest.from_dict(manifest)
if "participants" in payload:
payload["participants"] = [
Participant.from_dict(p) if isinstance(p, dict) else p for p in payload["participants"]
]
state = payload.get("state")
if isinstance(state, str):
payload["state"] = SessionState(state)
return cls(**payload)
|
Source code in autogen/beta/network/session.py
| def participant_ids(self) -> list[str]:
return [p.agent_id for p in self.participants]
|
Source code in autogen/beta/network/session.py
| def is_terminal(self) -> bool:
return is_terminal_session_state(self.state)
|