Events (v2)

magick_mind.resources.v2.events

Typed streaming events for the Cortex v2 Reason API.

@dataclass(frozen=True)
class ReasonEvent()

Base class for a typed Reason stream event.

@property
def content() -> str

Token content for token events, otherwise an empty string.

@dataclass(frozen=True)
class ReasonTokenEvent(ReasonEvent)

Incremental answer token/chunk.

@dataclass(frozen=True)
class ReasonCompleteEvent(ReasonEvent)

Terminal success event.

@dataclass(frozen=True)
class ReasonFailedEvent(ReasonEvent)

Terminal failure event.

@dataclass(frozen=True)
class ReasonThinkingEvent(ReasonEvent)

Progress or trace event suitable for thinking/progress UIs.

def parse_reason_event(event_type: str, data: dict[str, Any]) -> ReasonEvent

Parse one SSE frame into a typed event object.

async def iter_sse_events(
        lines: AsyncIterator[str]) -> AsyncIterator[ReasonEvent]

Yield typed events from an async iterator of SSE lines.

def parse_sse_text(text: str) -> list[ReasonEvent]

Parse a complete SSE payload. Mainly useful for tests.