Function process_external_message

methods/MIRIX/mirix/queue/__init__.py:75–115

Process a message consumed by an external system (e.g., Numaflow, custom Kafka consumer). This is the primary high-level API for integrating with external Kafka consumers or event processing systems. It handles all internal details of deserialization and processing. Args:

(raw_message: bytes)

async def process_external_message(raw_message: bytes) -> None:
    """
    Process a message consumed by an external system (e.g., Numaflow, custom Kafka consumer).

    This is the primary high-level API for integrating with external Kafka consumers or event
    processing systems. It handles all internal details of deserialization and processing.

    Args:
        raw_message: Raw message bytes from Kafka or event bus (JSON or protobuf format)

    Raises:
        ValueError: If message parsing fails
    """
    if not _manager.is_initialized:
        logger.info("Queue not initialized, auto-initializing with server for external message processing")
        from mirix.server.server import AsyncServer

        server = AsyncServer()
        await _manager.initialize(server=server)
        logger.info("Queue initialized with server instance")

    workers = _manager._workers
    if not workers:
        logger.error("No workers available after initialization - this should not happen!")
        raise RuntimeError("Failed to create queue workers during initialization")

    worker = workers[0]

    from mirix.queue.config import KAFKA_SERIALIZATION_FORMAT
    from mirix.queue.queue_util import deserialize_queue_message

    queue_message = deserialize_queue_message(raw_message, format=KAFKA_SERIALIZATION_FORMAT)

    logger.debug(
        "Processing external message (%s format): agent_id=%s, user_id=%s",
        KAFKA_SERIALIZATION_FORMAT,
        queue_message.agent_id,
        queue_message.user_id if queue_message.HasField("user_id") else "None",
    )

    await worker.process_external_message(queue_message)


__all__ = ["initialize_queue", "save", "process_external_message", "QueueMessage"]