kaboo_workflows.types¶

Shared event protocol and session manifest models.

kaboo_workflows.types ¶

Shared types for the core package.

This module is the single canonical home for cross-package types:

Node — alias for Agent | MultiAgentBase, the kinds of nodes that participate in a session.
EventType and StreamEvent — the typed-event protocol used by the event queue and external consumers.
The SessionManifest family of Pydantic models — the schema describing a wired session at invocation time. The manifest schema is intentionally decoupled from the YAML config schema in config.schema.

SessionManagerDescriptor `module-attribute` ¶

SessionManagerDescriptor = Annotated[
    FileProviderDescriptor
    | S3ProviderDescriptor
    | AgentCoreProviderDescriptor
    | CustomProviderDescriptor,
    Field(discriminator="provider"),
]

Discriminated union of session manager descriptors by provider type.

EventType ¶

Bases: StrEnum

Event type constants for :class:StreamEvent.

StrEnum values are plain strings, so == "token" comparisons work unchanged.

StreamEvent `dataclass` ¶

StreamEvent(
    type,
    agent_name,
    timestamp=(lambda: datetime.now(tz=(timezone.utc)))(),
    data=dict(),
)

A typed event from agent or multi-agent execution.

Per-agent activity (AGENT_START, TOKEN, REASONING, TOOL_START, TOOL_END, INTERRUPT, AGENT_COMPLETE, ERROR, NODE_START, NODE_STOP, HANDOFF, MULTIAGENT_START, MULTIAGENT_COMPLETE) is produced by :class:~kaboo_workflows.hooks.EventPublisher. Session-level events (SESSION_START, SESSION_END) are produced by the queue/wiring layer in :mod:kaboo_workflows.wire.

Attributes:

Name	Type	Description
`type`	`str`	Event type identifier (one of the :class:`EventType` values).
`agent_name`	`str`	Name of the agent or session entry point that produced this event.
`timestamp`	`datetime`	When the event occurred.
`data`	`dict[str, Any]`	Event-specific payload.

asdict ¶

asdict()

Convert this StreamEvent to a flat dict for serialization.

Source code in src/kaboo_workflows/types.py

def asdict(self) -> dict[str, Any]:
    """Convert this StreamEvent to a flat dict for serialization."""
    data = asdict(self)
    data["timestamp"] = self.timestamp.isoformat()
    return data

from_dict `classmethod` ¶

from_dict(data)

Deserialize a dict into a StreamEvent.

Restores all fields produced by :meth:asdict, including the timestamp (parsed from its ISO-8601 string representation).

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	A dict as produced by :meth:`asdict`.	required

Returns:

Type	Description
`StreamEvent`	A new StreamEvent instance.

Source code in src/kaboo_workflows/types.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> StreamEvent:
    """Deserialize a dict into a StreamEvent.

    Restores all fields produced by :meth:`asdict`, including the
    ``timestamp`` (parsed from its ISO-8601 string representation).

    Args:
        data: A dict as produced by :meth:`asdict`.

    Returns:
        A new StreamEvent instance.
    """
    ts_raw = data.get("timestamp")
    if isinstance(ts_raw, str):
        ts = datetime.fromisoformat(ts_raw)
    elif isinstance(ts_raw, datetime):
        ts = ts_raw
    else:
        ts = datetime.now(tz=timezone.utc)

    return cls(
        type=data.get("type", ""),
        agent_name=data.get("agent_name", ""),
        timestamp=ts,
        data=data.get("data", {}),
    )

eq ¶

__eq__(other)

Compare events by type, agent_name, and data (ignoring timestamp).

Source code in src/kaboo_workflows/types.py

def __eq__(self, other: object) -> bool:
    """Compare events by type, agent_name, and data (ignoring timestamp)."""
    if not isinstance(other, StreamEvent):
        return NotImplemented
    return (self.type, self.agent_name, self.data) == (other.type, other.agent_name, other.data)

hash ¶

__hash__()

Hash based on type and agent_name (data is unhashable).

Source code in src/kaboo_workflows/types.py

def __hash__(self) -> int:
    """Hash based on type and agent_name (data is unhashable)."""
    return hash((self.type, self.agent_name))

NodeRef ¶

Bases: BaseModel

Reference to a node in an orchestration topology.

Attributes:

Name	Type	Description
`id`	`str`	The node identifier (node_id for swarm/graph nodes).
`kind`	`str`	The node kind ("agent" or "orchestration").

EdgeRef ¶

Bases: BaseModel

Reference to a directed edge in a graph orchestration.

Attributes:

Name	Type	Description
`from_id`	`str`	The source node identifier.
`to_id`	`str`	The target node identifier.

ModelDescriptor ¶

Bases: BaseModel

Descriptor for an agent's model and provider.

Attributes:

Name	Type	Description
`model_id`	`str \| None`	The model identifier (e.g. "us.anthropic.claude-sonnet-4-6"), or None if not available.
`provider`	`str`	The fully-qualified class name of the model provider.

FileProviderDescriptor ¶

Bases: BaseModel

Session manager descriptor for file-based storage.

Attributes:

Name	Type	Description
`provider`	`Literal['file']`	Literal "file".
`session_id`	`str`	The session identifier.
`storage_dir`	`str`	The filesystem directory where sessions are stored.

S3ProviderDescriptor ¶

Bases: BaseModel

Session manager descriptor for S3-based storage.

Attributes:

Name	Type	Description
`provider`	`Literal['s3']`	Literal "s3".
`session_id`	`str`	The session identifier.
`bucket`	`str`	The S3 bucket name.
`prefix`	`str`	The S3 key prefix (empty string if no prefix).

AgentCoreProviderDescriptor ¶

Bases: BaseModel

Session manager descriptor for AgentCore Memory storage.

Attributes:

Name	Type	Description
`provider`	`Literal['agentcore']`	Literal "agentcore".
`session_id`	`str`	The session identifier.
`memory_id`	`str`	The AgentCore memory identifier.
`actor_id`	`str`	The AgentCore actor identifier.

CustomProviderDescriptor ¶

Bases: BaseModel

Session manager descriptor for custom session manager implementations.

Attributes:

Name	Type	Description
`provider`	`Literal['custom']`	Literal "custom".
`session_id`	`str \| None`	The session identifier, or None if not available.
`class_name`	`str`	The fully-qualified class name of the session manager.

AgentDescriptor ¶

Bases: BaseModel

Descriptor for a configured agent in the session.

Attributes:

Name	Type	Description
`name`	`str`	The agent's configured name.
`description`	`str \| None`	The agent's description, or None.
`model`	`ModelDescriptor`	The agent's model descriptor.
`session_manager`	`SessionManagerDescriptor \| None`	The agent's session manager descriptor, or None.

OrchestrationDescriptor ¶

Bases: BaseModel

Descriptor for a configured orchestration in the session.

Attributes:

Name	Type	Description
`name`	`str`	The orchestration's configured name.
`kind`	`str`	The orchestration kind ("delegate", "swarm", "graph", or "unknown").
`session_manager`	`SessionManagerDescriptor \| None`	The orchestration's session manager descriptor, or None.
`nodes`	`list[NodeRef]`	List of nodes in the orchestration topology.
`edges`	`list[EdgeRef] \| None`	List of edges in the orchestration topology (None for swarm/delegate).
`entry_node_id`	`str \| None`	The entry node identifier(s), or None.

EntryDescriptor ¶

Bases: BaseModel

Descriptor identifying the session entry point.

Attributes:

Name	Type	Description
`name`	`str`	The entry point's configured name.
`kind`	`str`	The entry point kind ("agent" or "orchestration").

SessionManifest ¶

Bases: BaseModel

Manifest describing the wired session topology and configuration.

Attributes:

Name	Type	Description
`agents`	`list[AgentDescriptor]`	List of agent descriptors.
`orchestrations`	`list[OrchestrationDescriptor]`	List of orchestration descriptors.
`entry`	`EntryDescriptor`	The entry point descriptor.

kaboo_workflows.types¶