EggAI SDK

EggAI Multi-Agent Meta Framework is an async-first framework for building, deploying, and scaling multi-agent systems for modern enterprise environments.

Features

• Multi-Transport Support: In-Memory, Redis Streams, Kafka, and extensible for custom transports
• Async-First: Built on asyncio for high-performance concurrent processing
• Type-Safe: Full type hints and Pydantic validation with CloudEvents-compliant message protocol
• Production-Ready: Comprehensive error handling, logging, and monitoring
• Flexible Architecture: Easily extensible with custom transports and middleware
• Protocol Interoperability: A2A for agent collaboration, MCP for LLM tools, CloudEvents for messaging

Installation

pip install eggai

Redis Streams and Kafka support are included by default.

Optional extras:

pip install eggai[otel]   # Distributed tracing via OpenTelemetry
pip install eggai[cli]    # CLI tools for scaffolding
pip install eggai[a2a]    # A2A (Agent-to-Agent) SDK integration
pip install eggai[mcp]    # MCP (Model Context Protocol) support

Quick Start

Using In-Memory Transport (Testing & Development)

Perfect for unit tests, prototyping, and local development.

from eggai import Agent, Channel
from eggai.transport.memory import InMemoryTransport

# Create an in-memory transport
transport = InMemoryTransport()

# Create an agent
agent = Agent(name="my-agent", transport=transport)

# Define message handler
@agent.subscribe(channel=Channel("input-channel", transport=transport))
async def handle_message(message):
    # Process message
    result = await process(message)
    # Publish to output channel
    output_channel = Channel("output-channel", transport=transport)
    await output_channel.publish(result)

# Start the agent
await agent.start()

Using Redis Streams (Recommended for Production)

Redis Streams provides durable message delivery with consumer groups, perfect for microservices and production deployments.

from eggai import Agent, Channel
from eggai.transport.redis import RedisTransport

# Create a Redis transport
transport = RedisTransport(url="redis://localhost:6379")

# Create an agent
agent = Agent(name="my-agent", transport=transport)

# Define message handler
@agent.subscribe(channel=Channel("input-channel", transport=transport))
async def handle_message(message):
    # Process message
    result = await process(message)
    # Publish to output channel
    output_channel = Channel("output-channel", transport=transport)
    await output_channel.publish(result)

# Start the agent
await agent.start()

Using Kafka (Production - High Throughput)

Kafka is ideal for high-throughput, large-scale distributed streaming workloads.

from eggai import Agent, Channel
from eggai.transport.kafka import KafkaTransport

# Create a Kafka transport
transport = KafkaTransport(
    bootstrap_servers="localhost:9092"
)

# Create an agent
agent = Agent(name="my-agent", transport=transport)

# Define message handler
@agent.subscribe(channel=Channel("input-topic", transport=transport))
async def handle_message(message):
    # Process message
    result = await process(message)
    # Publish to output topic
    output_channel = Channel("output-topic", transport=transport)
    await output_channel.publish(result)

# Start the agent
await agent.start()

Transport Comparison

Reliable Message Delivery (Redis Streams)

Pending Entries List (PEL) and Retry Streams

With NACK_ON_ERROR (the default), a handler exception leaves the message in Redis's Pending Entries List (PEL). FastStream only reads new messages, so a failed message would be stuck forever without intervention.

Enable SDK-managed retry by setting retry_on_idle_ms on any subscription:

from eggai import Agent, Channel
from eggai.transport import RedisTransport

transport = RedisTransport(url="redis://localhost:6379")
agent = Agent("order-service", transport=transport)
orders = Channel("orders", transport=transport)

@agent.subscribe(
    channel=orders,
    retry_on_idle_ms=30_000,          # reclaim after 30s idle (required to enable retry)
    max_retries=5,                    # route to DLQ after 5 retries (default: 5, None = unlimited)
    retry_reclaim_interval_s=15.0,    # PEL scan interval (default: 15.0)
    retry_backoff_multiplier=2.0,     # exponential backoff between retries (default: 1.0 = constant)
    retry_backoff_max_ms=900_000,     # cap the escalated delay (default: None = uncapped)
    retry_backoff_jitter=0.2,         # add up to 20% on top of each delay to avoid a thundering herd (default: 0.0)
    on_dlq=None,                      # optional async/sync callback(fields, msg_id, count)
)
async def handle_order(message):
    # If this raises, the message stays in the PEL.
    # After 30s idle the reclaimer moves it to a per-handler retry stream
    # (e.g. eggai.orders.order-service-handle_order-1.retry) and this same
    # handler is called again. After 5 failed retries, the message is routed
    # to the per-handler DLQ (eggai.orders.order-service-handle_order-1.dlq).
    await process_order(message)

await agent.start()

The SDK automatically:

1. Starts a background reclaimer that scans the PEL every 15 seconds (configurable via retry_reclaim_interval_s).
2. Moves idle messages (older than retry_on_idle_ms) to a dedicated per-handler {channel}.{handler_suffix}.retry stream. The per-handler suffix keeps one handler's failures from being broadcast to other handlers on the same channel.
3. Subscribes the same handler to the retry stream.
4. Runs a second reclaimer on the retry stream that re-queues back to itself (no .retry.retry chain).
5. After max_retries retry attempts (default 5), routes the message to a per-handler {channel}.{handler_suffix}.dlq Dead Letter Queue instead of retrying again.

Delivery guarantee: at-least-once. XADD and XACK are not atomic — a crash between them will re-deliver the message on the next cycle. Handlers must be idempotent. Two fields are injected on retry delivery to aid deduplication:

Dead Letter Queue (DLQ): Messages that exceed max_retries (default 5) are routed to the per-handler {channel}.{handler_suffix}.dlq. The DLQ is terminal — no automatic reclaimer. Set max_retries=None for unlimited retries. An optional on_dlq callback fires when a message lands in the DLQ.

Retry backoff: By default retries fire at a constant cadence equal to retry_on_idle_ms. Set retry_backoff_multiplier > 1.0 to back off exponentially: a message is treated as due for reclaim once it has been idle for retry_on_idle_ms * (retry_backoff_multiplier ** _retry_count), capped at retry_backoff_max_ms. With a 30s base and multiplier 2.0 the attempts space out as 30s → 60s → 120s → 240s …, giving an overloaded or recovering downstream room to breathe instead of being retried on a fixed clock. retry_backoff_jitter (a fraction in [0, 1]) adds a random 0..jitter slice on top of each computed threshold so a fleet of workers that failed during the same outage don't come due in lockstep. Jitter is applied upward, never below retry_on_idle_ms — that base is a hard floor, since the reclaimer won't touch a message idle for less than it (it may be slow rather than stuck). Note the spread is only effective once jitter × threshold is comparable to retry_reclaim_interval_s (the scan cadence quantises smaller jitter away). Backoff is opt-in and backward compatible — the default multiplier=1.0 preserves the constant retry_on_idle_ms spacing exactly.

Automatic recovery from Redis stream loss (NOGROUP): If Redis loses streams (restart without persistence, failover, memory eviction), the SDK auto-recovers. A background monitor periodically ensures consumer groups exist via XGROUP CREATE with MKSTREAM, and the reclaimer recreates groups on NOGROUP errors. No configuration needed — always active with RedisTransport.

Constraints:

• min_idle_time (FastStream XAUTOCLAIM) and retry_on_idle_ms are mutually exclusive on the same subscription — mixing them raises ValueError.
• Binary (non-UTF-8) field values are not supported; use JSON-serialisable payloads.

Retry delivery reference

Main stream  (eggai.orders)
    │
    ├── FastStream consumer  (group/consumer: order-service-handle_order-1)
    │       on exception → NACK → message stays in main PEL
    │
    └── Reclaimer            (consumer: order-service-handle_order-1-reclaimer)
            every 15s: XPENDING → idle > 30s → XCLAIM
            _retry_count ≤ max_retries → XADD <retry stream> → XACK
            _retry_count > max_retries → XADD <dlq stream>   → XACK

Retry stream (eggai.orders.order-service-handle_order-1.retry)   ← per-handler
    │
    ├── FastStream consumer  (group: order-service-handle_order-1-retry)
    │       same handler — on exception → NACK → message stays in retry PEL
    │
    └── Reclaimer            (target: same retry stream — no .retry.retry chain)
            every 15s: XPENDING → idle > 30s → XCLAIM
            _retry_count ≤ max_retries → XADD <retry stream> → XACK
            _retry_count > max_retries → XADD <dlq stream>   → XACK

DLQ stream   (eggai.orders.order-service-handle_order-1.dlq)      ← per-handler
        terminal — no reclaimer, manual re-drive only

Observability (OpenTelemetry)

EggAI has built-in distributed tracing via OpenTelemetry. Every message hop — from publisher through to handler — shares a single trace_id, giving you end-to-end visibility across agents.

Install the optional tracing dependencies:

pip install eggai[otel]

Auto-configuration via environment variables

If OTEL_EXPORTER_OTLP_ENDPOINT is set when the process starts, tracing activates automatically. The protocol is selected from OTEL_EXPORTER_OTLP_PROTOCOL (defaults to grpc):

# gRPC backend (Jaeger, Grafana Tempo, …)
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317
export OTEL_EXPORTER_OTLP_PROTOCOL=grpc
export OTEL_SERVICE_NAME=my-agent

# HTTP backend (Langfuse, Honeycomb, Datadog, …)
export OTEL_EXPORTER_OTLP_ENDPOINT=https://cloud.langfuse.com/api/public/otel
export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic <base64-encoded-key>"
export OTEL_SERVICE_NAME=my-agent

No code changes needed — just set the env vars and run your agent.

Explicit setup in code

from eggai import setup_tracing, Agent, Channel

# gRPC (default)
setup_tracing(service_name="my-agent")

# HTTP (required for Langfuse, Honeycomb, Datadog, etc.)
setup_tracing(exporter="otlp-http", service_name="my-agent")

# Console (useful during development)
setup_tracing(exporter="console", service_name="my-agent")

Call setup_tracing() once at startup, before any agents publish or subscribe.

What gets traced automatically

Every channel.publish() creates a producer span and every handler invocation creates a consumer span. All spans within a single request share the same trace_id:

eggai.publish eggai.orders          ← producer span (your publish call)
  └─ eggai.process eggai.orders     ← consumer span (handler invocation)
       └─ eggai.publish eggai.bills ← producer span (handler publishes downstream)
            └─ eggai.process …      ← and so on

Span attributes set on every span:

Langfuse

Langfuse supports OTLP natively (HTTP only, v3.22+). Point EggAI at its endpoint and your agent traces appear as Langfuse traces with each span as an observation:

export OTEL_EXPORTER_OTLP_ENDPOINT=https://cloud.langfuse.com/api/public/otel
export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
export OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic $(echo -n 'pk-lf-...:sk-lf-...' | base64)"
export OTEL_SERVICE_NAME=my-agent

To attach a Langfuse session or user to spans, set the attributes on your own root span before publishing:

from opentelemetry import trace

tracer = trace.get_tracer("myapp")

with tracer.start_as_current_span("handle-user-request") as span:
    span.set_attribute("langfuse.session.id", "session-abc")
    span.set_attribute("langfuse.user.id", "user-123")
    await channel.publish(msg)  # EggAI spans are nested under this span

Self-hosted Langfuse: replace the endpoint with http://your-host:3000/api/public/otel.

Production Recommendations

For production deployments, we recommend:

• Redis Streams: Best for most production workloads. Simpler to operate, lower cost, excellent performance for microservices and event-driven architectures.
• Kafka: Best for very high-throughput requirements (1M+ messages/sec), complex stream processing, or when you need advanced features like exactly-once semantics.

The In-Memory transport should only be used for testing and development.

Interoperability

EggAI is designed as a meta-framework that enables seamless integration with other agent systems and protocols:

Agent-to-Agent (A2A) Protocol

Connect EggAI agents with other agent frameworks and systems using the A2A protocol:

pip install eggai[a2a]

A2A enables:

• Cross-framework agent communication
• Standardized message formats for multi-agent systems
• Integration with existing agent ecosystems (AutoGen, LangChain, etc.)
• HTTP-based service discovery via AgentCards

Read full A2A documentation →

Model Context Protocol (MCP)

Integrate with LLM tools and services through MCP:

pip install eggai[mcp]

MCP support enables:

• LLM tool calling and function execution
• Context sharing across agent conversations
• Integration with Claude, GPT, and other LLM providers
• Standardized tool interfaces via FastMCP

Read full MCP documentation →

Transport Flexibility

EggAI's transport abstraction allows agents to communicate regardless of underlying infrastructure:

• Hybrid deployments: Mix local (in-memory) and distributed (Redis/Kafka) agents
• Migration paths: Start with Redis, scale to Kafka without changing agent code
• Multi-cloud: Deploy agents across different cloud providers using different transports
• Custom transports: Extend with your own transport implementations

Extensibility

Build custom integrations through:

• Custom Transport API: Implement your own message broker backends
• Middleware system: Add cross-cutting concerns (logging, metrics, auth)
• Message adapters: Transform messages between different protocols
• Plugin architecture: Extend agent capabilities with reusable components

Documentation

Core Documentation

• Message Protocol: CloudEvents-based message structure for agent communication
• Agent: Core agent abstraction for building autonomous units
• Channel: Communication layer for event publishing and subscription

Additional Resources

For full documentation, visit: https://docs.egg-ai.com/

License

MIT License - see LICENSE file for details