magick_mind.realtime.handler

Decorator-based event router for Centrifugo realtime publications.

Usage: router = EventRouter()

Event-only handler (backward compatible)

@router.on("chat_message") async def handle_chat(event: ChatMessageEvent): print(event.payload.message)

Handler with EventContext — identifies which end-user the event is for

@router.on("chat_message") async def handle_chat(event: ChatMessageEvent, ctx: EventContext): print(f"Message for {ctx.target_user_id}: {event.payload.message}")

@router.on("image_generation") async def handle_image(event: ImageGenerationEvent, ctx: EventContext): print(f"Image for {ctx.target_user_id}: {event.payload.data}")

Catch-all for unregistered event types

@router.on_unknown async def handle_unknown(event: UnknownEvent, ctx: EventContext): logger.warning(f"Unhandled event type: {event.type} on {ctx.channel}")

EventRouter

class EventRouter(ClientEventHandler)

Routes Centrifugo publications to registered async callbacks by event type.

Integrates with centrifuge-python's ClientEventHandler so it can be passed directly to RealtimeClient.connect().

Handlers may accept one or two positional arguments::

@router.on("chat_message") async def handle(event: ChatMessageEvent): ... # event only

@router.on("chat_message") async def handle(event: ChatMessageEvent, ctx: EventContext): ... # with context

EventRouter.on

def on(event_type: str) -> Callable[[EventCallback], EventCallback]

Register a handler for a specific event type.

Arguments:

• event_type - The WsEvent type string (e.g. "chat_message", "image_generation")

Returns:

Decorator that registers the handler function.

EventRouter.on_unknown

@property
def on_unknown() -> Callable[[EventCallback], EventCallback]

Register a catch-all handler for unknown/unregistered event types.

Usage: @router.on_unknown async def handle_unknown(event: UnknownEvent): ...

@router.on_unknown async def handle_unknown(event: UnknownEvent, ctx: EventContext): ...

EventRouter.on_publication

async def on_publication(ctx: ServerPublicationContext) -> None

Internal: called by centrifuge-python on each publication. Parses raw data into typed event, dispatches to registered handler.

magick_mind.realtime.events

Realtime event models for Centrifugo publications from Xavier.

Xavier source of truth: services/xavier/internal/types/broadcast.go

Usage: from magick_mind.realtime.events import parse_ws_event, ChatMessageEvent

event = parse_ws_event(ctx.pub.data) match event: case ChatMessageEvent(payload=p): handle_chat(p) case ImageGenerationEvent(payload=p): handle_artifact(p) case UnknownEvent(type=t): logger.warning(f"unhandled: {t}")

EventContext

@dataclass(frozen=True, slots=True)
class EventContext()

SDK-derived metadata about where a realtime event came from.

Attributes:

• channel - Raw Centrifugo channel string.
• target_user_id - The end-user ID extracted from the channel (personal:<target>#<service>).

EventContext.from_channel

@classmethod
def from_channel(cls, channel: str) -> EventContext

Parse a personal:{target}#{service} channel string.

ChatMessagePayload

class ChatMessagePayload(BaseModel)

Payload for type="chat_message" realtime events.

ChatMessageEvent

class ChatMessageEvent(BaseModel)

Realtime event: AI chat response.

ArtifactData

class ArtifactData(BaseModel)

Artifact schema for Centrifugo broadcast.

ArtifactPayload

class ArtifactPayload(BaseModel)

Payload for type="image_generation" realtime events.

ImageGenerationEvent

class ImageGenerationEvent(BaseModel)

Realtime event: artifact/image generated.

UnknownEvent

class UnknownEvent(BaseModel)

Catch-all for unrecognised event types.

parse_ws_event

def parse_ws_event(
    data: dict[str, Any]
) -> ChatMessageEvent | ImageGenerationEvent | UnknownEvent

Parse raw Centrifugo publication data into a typed event.

Dict dispatch — O(1), no try/except, no Pydantic discriminator magic. Unknown types degrade gracefully to UnknownEvent.

magick_mind.realtime.client

Realtime client implementation using centrifuge-python.

_DelegatingSubscriptionHandler

class _DelegatingSubscriptionHandler(SubscriptionEventHandler)

Routes client-side subscription publications to EventRouter.on_publication.

_DelegatingSubscriptionHandler.on_publication

async def on_publication(ctx: PublicationContext) -> None

Route client-side publication to the EventRouter.

RealtimeClient

class RealtimeClient()

Async client for real-time features using WebSockets. Uses pure client-side subscriptions for reliability.

RealtimeClient.on

def on(event_type: str) -> Callable[[EventCallback], EventCallback]

Register a handler for a realtime event type.

RealtimeClient.on_unknown

@property
def on_unknown() -> Callable[[EventCallback], EventCallback]

Register a catch-all for unknown event types.

RealtimeClient.connect

async def connect() -> None

Connect to the realtime service.

RealtimeClient.disconnect

async def disconnect() -> None

Disconnect from the realtime service.

RealtimeClient.subscribe

async def subscribe(target_user_id: str) -> None

Subscribe to a user's channel using client-side subscription.

Arguments:

• target_user_id - ID of the user to subscribe to

RealtimeClient.subscribe_many

async def subscribe_many(target_user_ids: List[str]) -> None

Subscribe to multiple users concurrently.

RealtimeClient.unsubscribe

async def unsubscribe(target_user_id: str) -> None

Unsubscribe from a user's channel.

RealtimeClient.unsubscribe_many

async def unsubscribe_many(target_user_ids: List[str]) -> None

Unsubscribe from multiple users concurrently.

RealtimeClient.client

@property
def client() -> Optional[Client]

Get underlying centrifuge client.