Hands-On: Enriched Chatbot

This example demonstrates how tinychat handles real-world complexity: a chatbot that enriches every user message with CRM data and conversation history before generating context-aware responses.

What We’re Building

A production-ready conversational system that:

Retrieves user information from a CRM system
Loads conversation history from a database
Routes enriched messages through an agent system
Generates personalized responses based on full context
Uses nested composites to organize logical subsystems

Message Types: Documenting the Flow

This example defines four custom message types that make the information flow explicit:

@dataclass(frozen=True)
class CRMMessage(Message):
    content: dict
    user_id: str
    user_info: dict
    user_message: Message

@dataclass(frozen=True)
class EnrichedMessage(Message):
    content: dict
    user_id: str
    user_info: dict
    history: list[str]
    user_message: Message

@dataclass(frozen=True)
class ReplyRequestMessage(Message):
    content: str
    user_id: str
    user_info: dict
    history: list[str]
    user_message: Message

@dataclass(frozen=True)
class ReplyMessage(Message):
    content: str
    user_message: Message

Why custom message types?

Each message type represents a distinct stage of enrichment:

IngressMessage → Raw user input
CRMMessage → Input + user profile
EnrichedMessage → Input + profile + history
ReplyRequestMessage → Formatted request for agent
ReplyMessage → Generated response
EgressMessage → Final output

The types create a type-driven pipeline where each processor knows exactly what data it receives and what data it must produce.

Immutability with accumulated state

Notice how each message type carries forward previous data:

CRMMessage includes user_message (the original ingress)
EnrichedMessage includes user_message AND user_info
ReplyRequestMessage includes everything

This is immutable accumulation—we’re not mutating messages, we’re creating new messages with more information. Each message is a complete snapshot of the processing state at that stage.

Self-documenting architecture

Just by reading the message type definitions, you understand the entire system:

# The progression tells the story:
IngressMessage          # → "User said something"
  → CRMMessage         # → "We know who they are"
    → EnrichedMessage  # → "We know their history"
      → ReplyRequest   # → "Agent needs to respond"
        → ReplyMessage # → "Agent generated response"
          → Egress     # → "Send to user"

No documentation needed. The types ARE the documentation.

Stage 1: CRM Enrichment

The CRMProcessor retrieves user profile information:

class CRMProcessor(MessageProcessor):
    def __init__(self, name: str = "CRM", output_types: set[type[Message]] | None = None):
        super().__init__(name=name, output_types=output_types)
        # Mock user database
        self._users = {
            "user_123": {
                "name": "Alice",
                "email": "[email protected]",
                "tier": "premium",
            },
        }
    
    async def _process(self, message: IngressMessage) -> Optional[Message]:
        user_id = message.user_id
        user_info = self._users.get(user_id, {"name": "Unknown", ...})
        
        return CRMMessage(
            content={"user_id": user_id, "user_info": user_info},
            user_id=user_id,
            user_message=message,
            user_info=user_info,
        )

The processor maintains internal state (the user database) but produces immutable messages. This separation means you can cache, replicate, or hot-swap the processor without worrying about message consistency.

Real-World Extensions

External CRM
With Caching
With Retry Logic

class CRMProcessor(MessageProcessor):
    def __init__(self, crm_client: CRMClient):
        self._crm = crm_client
    
    async def _process(self, message: IngressMessage) -> Optional[Message]:
        # Call external CRM API
        user_info = await self._crm.get_user(message.user_id)
        
        if not user_info:
            return None  # Filter out unknown users
        
        return CRMMessage(
            user_id=message.user_id,
            user_info=user_info,
            user_message=message,
        )

class CachedCRMProcessor(MessageProcessor):
    def __init__(self, crm_client: CRMClient, cache: Cache):
        self._crm = crm_client
        self._cache = cache
    
    async def _process(self, message: IngressMessage) -> Optional[Message]:
        # Check cache first
        cached = await self._cache.get(f"user:{message.user_id}")
        if cached:
            return CRMMessage(user_info=cached, ...)
        
        # Fetch and cache
        user_info = await self._crm.get_user(message.user_id)
        await self._cache.set(f"user:{message.user_id}", user_info)
        
        return CRMMessage(user_info=user_info, ...)

class ReliableCRMProcessor(MessageProcessor):
    async def _process(self, message: IngressMessage) -> Optional[Message]:
        for attempt in range(3):
            try:
                user_info = await self._crm.get_user(message.user_id)
                return CRMMessage(user_info=user_info, ...)
            except CRMAPIError as e:
                if attempt == 2:
                    logger.error(f"CRM failed after 3 attempts: {e}")
                    return None
                await asyncio.sleep(2 ** attempt)

Notice how policies like caching and retry logic are your choice, not framework requirements. Implement what makes sense for your system.

Stage 2: Database Enrichment

The DBProcessor adds conversation history:

class DBProcessor(MessageProcessor):
    def __init__(self, name: str = "DB", output_types: set[type[Message]] | None = None):
        super().__init__(name=name, output_types=output_types)
        # Mock chat history database
        self._history = {
            "user_123": [
                "User: Hello!",
                "Bot: Hi Alice! How can I help?",
                ...
            ],
        }
    
    async def _process(self, message: CRMMessage) -> Optional[Message]:
        user_id = message.user_id
        history = self._history.get(user_id, [])
        
        return EnrichedMessage(
            content={"user_id": user_id, "user_info": message.user_info, "history": history},
            user_id=user_id,
            user_info=message.user_info,
            history=history,
            user_message=message.user_message,
        )

The DB processor receives CRMMessage and produces EnrichedMessage. The type system ensures we can’t accidentally skip the CRM stage—the types enforce the enrichment order.

Automatic Chaining

Notice we didn’t write any code to chain CRM → DB. The CompositeProcessor handles it automatically:

bus = CompositeProcessor(
    handlers={
        IngressMessage: crm,   # Produces CRMMessage
        CRMMessage: db,        # Produces EnrichedMessage
        EnrichedMessage: agent_bus,
    },
)

# This automatically chains:
# Ingress → CRM → CRMMessage → DB → EnrichedMessage → agent_bus

The routing is type-driven and automatic. No manual wiring, no callbacks, no complex orchestration code.

Stage 3: Agent System (Nested Composite)

Here’s where it gets interesting. We create a nested composite to handle the agent logic:

# Inner message bus for agents
agent_bus = CompositeProcessor(
    handlers={
        EnrichedMessage: orchestrator,      # Routes to reply agent
        ReplyRequestMessage: reply_agent,   # Generates response
        ReplyMessage: orchestrator,         # Approves and returns
    },
    max_hops=10,
    output_types={EgressMessage},
)

# Outer message bus
bus = CompositeProcessor(
    handlers={
        IngressMessage: crm,
        CRMMessage: db,
        EnrichedMessage: agent_bus,  # Nested composite!
    },
    max_hops=5,
    output_types={EgressMessage},
)

Why nest composites?

Nesting creates logical subsystems with clear boundaries:

Outer bus: Handles enrichment (CRM → DB)
Inner bus: Handles agent logic (Orchestrator → Reply → Orchestrator)

Each subsystem has its own max_hops limit, error handling, and lifecycle. The outer bus doesn’t need to know about the inner agent routing—it just knows “send EnrichedMessage to agent system, get back EgressMessage.”

CompositeProcessor is a MessageProcessor

This is the key insight: CompositeProcessor implements MessageProcessor interface. That means a composite can be a handler in another composite.

# agent_bus is a MessageProcessor
handlers={
    EnrichedMessage: agent_bus,  # It's just another processor!
}

This is recursive composition—the same abstraction at every level.

Independent hop limits

agent_bus = CompositeProcessor(max_hops=10)  # Inner limit
bus = CompositeProcessor(max_hops=5)         # Outer limit

The inner agent bus can do up to 10 message transformations internally. The outer bus can do up to 5 transformations in its own routing. These limits are independent and composable.

The Orchestrator: State-Dependent Routing (1:N)

The OrchestratorAgent demonstrates the 1:N handler-to-types pattern:

class OrchestratorAgent(MessageProcessor):
    async def _process(self, message: Message) -> Optional[Message]:
        if isinstance(message, EnrichedMessage):
            # Forward to reply agent
            return ReplyRequestMessage(
                content=f"New message from {message.user_id}: {message.user_message.content}",
                user_id=message.user_id,
                user_info=message.user_info,
                history=message.history,
                user_message=message.user_message,
            )
        
        elif isinstance(message, ReplyMessage):
            # Approve and return as egress
            return EgressMessage(
                content=message.content,
            )
        
        else:
            raise ValueError(f"Unexpected message type: {type(message)}")

The orchestrator handles two different input types and produces two different output types. This is what enables complex routing topologies.

Flow Through the Agent Bus

Let’s trace a message through the agent bus:

EnrichedMessage arrives

The outer bus routes EnrichedMessage to the agent bus.

# agent_bus receives EnrichedMessage

Orchestrator transforms to ReplyRequest

The orchestrator receives EnrichedMessage and transforms it to ReplyRequestMessage.

EnrichedMessage → Orchestrator → ReplyRequestMessage

ReplyAgent generates response

The agent bus routes ReplyRequestMessage to the reply agent, which produces ReplyMessage.

ReplyRequestMessage → ReplyAgent → ReplyMessage

Orchestrator approves and returns

The agent bus routes ReplyMessage back to the orchestrator, which approves and produces EgressMessage.

ReplyMessage → Orchestrator → EgressMessage

Agent bus returns to outer bus

The agent bus returns EgressMessage to the outer bus, which completes processing.

Three message transformations within the agent bus, all automatic. The outer bus just sees “sent EnrichedMessage, got back EgressMessage.”

Why This Pattern?

Separation of Concerns

Orchestrator = routing logic. ReplyAgent = generation logic. Each processor does one thing.

Easy to Extend

Need approval logic? Add an approval processor. Need multiple agents? Add more handlers.

Testable in Isolation

Test orchestrator routing separately from reply generation. Mock agents, test flows.

Composable State Machines

The orchestrator can route to different processors based on any logic—user tier, message type, business rules.

The Reply Agent: Context-Aware Generation

The ReplyAgent uses all the enriched context:

class ReplyAgent(MessageProcessor):
    async def _process(self, message: ReplyRequestMessage) -> Optional[Message]:
        # All context available in the message
        user_name = message.user_info.get("name", "there")
        tier = message.user_info.get("tier", "free")
        history_count = len(message.history)
        
        # Generate personalized response
        reply_text = (
            f"Hello {user_name}! Thanks for your message: '{message.user_message.content}'. "
            f"I can see you're a {tier} member and we've chatted {history_count} times before. "
            f"How can I assist you today?"
        )
        
        return ReplyMessage(
            content=reply_text,
            user_message=message.user_message,
        )

The reply agent doesn’t need to know about CRM processors or database processors. It just receives a ReplyRequestMessage with all the context it needs. This is message-driven decoupling.

Swapping Implementations

Want to use an LLM instead of template-based replies?

class LLMReplyAgent(MessageProcessor):
    def __init__(self, llm_client: LLMClient):
        self._llm = llm_client
    
    async def _process(self, message: ReplyRequestMessage) -> Optional[Message]:
        # Build prompt with enriched context
        prompt = f"""
        User: {message.user_info['name']} (tier: {message.user_info['tier']})
        History: {message.history}
        Current message: {message.user_message.content}
        
        Generate a helpful, personalized response.
        """
        
        response = await self._llm.generate(prompt)
        
        return ReplyMessage(
            content=response,
            user_message=message.user_message,
        )

Just swap the handler in the composite—the rest of the system doesn’t change. The orchestrator doesn’t care how replies are generated, only that ReplyRequestMessage → ReplyMessage.

Observability: Tracking the Journey

The LoggingObserver tracks each message transformation with latency metrics:

class LoggingObserver(BaseObserver):
    def __init__(self):
        self._received_messages: dict[str, MessageReceived] = {}
    
    async def on_message_received(self, message: MessageReceived) -> None:
        logger.info(f"📨 [{message.source_processor.name}] Received: {message.content}")
        self._received_messages[message.source_message.id] = message
    
    async def on_message_processed(self, message: MessageProcessed) -> None:
        logger.info(f"✅ [{message.source_processor.name}] Processed: {message.content}")
        
        # Calculate latency
        received_msg = self._received_messages.get(message.source_message.id)
        if received_msg:
            latency_ns = message.timestamp - received_msg.timestamp
            latency_us = latency_ns / 1_000
            logger.info(f"⏱️  [{message.source_processor.name}] Latency: {latency_us:.2f}μs")

Observers see everything

The observer receives notifications for:

Every message received by every processor
Every message processed by every processor
Every exception that occurs

This gives you complete visibility into the system without touching processor logic.

Calculate cross-cutting metrics

The observer tracks message IDs to calculate per-processor latency. You could also calculate:

End-to-end latency (Ingress → Egress)
Throughput (messages/second per processor)
Error rates (exceptions/total messages)
Hop counts (messages hitting max_hops)

All without modifying any processor code.

Multiple observers

config = SetupConfig(
    observers=[
        LoggingObserver(),
        MetricsObserver(),
        TracingObserver(),
        DebugObserver(),
    ],
)

Each observer handles different concerns. They don’t interfere with each other or with processing logic.

Lifecycle: Assembly and Execution

The main flow demonstrates the full lifecycle:

async def main():
    # 1. Configuration
    config = SetupConfig(
        task_manager_params=TaskManagerParams(loop=asyncio.get_running_loop()),
        observers=[LoggingObserver()],
    )
    
    # 2. Create processors with output types
    crm = CRMProcessor(output_types={CRMMessage})
    db = DBProcessor(output_types={EnrichedMessage})
    orchestrator = OrchestratorAgent(
        name="orchestrator",
        output_types={ReplyRequestMessage, EgressMessage}  # 1:N!
    )
    reply_agent = ReplyAgent(name="reply_agent", output_types={ReplyMessage})
    
    # 3. Build nested topology
    agent_bus = CompositeProcessor(
        handlers={
            EnrichedMessage: orchestrator,
            ReplyRequestMessage: reply_agent,
            ReplyMessage: orchestrator,
        },
        max_hops=10,
        output_types={EgressMessage},
    )
    
    bus = CompositeProcessor(
        handlers={
            IngressMessage: crm,
            CRMMessage: db,
            EnrichedMessage: agent_bus,  # Nested!
        },
        max_hops=5,
        output_types={EgressMessage},
    )
    
    # 4. Setup
    await bus.setup(config)
    
    # 5. Process
    message = IngressMessage(
        content="I need help with my account",
        conversation_id="enriched-demo",
        user_id="user_123",
    )
    
    result = await bus.process(message)

Output type validation happens at composite construction. If you declare that a processor outputs MessageType but no handler exists for MessageType, you’ll get an error immediately—not at runtime.

Topology Visualization

The system’s structure is self-documenting:

# Get topology for debugging
topology = bus.get_topology()
# Shows:
# IngressMessage → CRMProcessor → CRMMessage
# CRMMessage → DBProcessor → EnrichedMessage  
# EnrichedMessage → CompositeProcessor (agent_bus) → EgressMessage
#   [Inside agent_bus]
#   EnrichedMessage → Orchestrator → ReplyRequestMessage
#   ReplyRequestMessage → ReplyAgent → ReplyMessage
#   ReplyMessage → Orchestrator → EgressMessage

Use get_topology() to generate documentation, debug routing issues, or visualize complex systems in dashboards.

Scaling Patterns

This architecture scales both vertically and horizontally:

Add More Enrichment
Add More Agents
Add Parallel Processing
Add Nested Systems

bus = CompositeProcessor(
    handlers={
        IngressMessage: crm,
        CRMMessage: preferences_processor,  # New stage
        PreferencesMessage: db,
        EnrichedMessage: sentiment_analyzer, # New stage
        AnalyzedMessage: agent_bus,
    },
)

Just add message types and handlers. The composite handles routing.

agent_bus = CompositeProcessor(
    handlers={
        EnrichedMessage: orchestrator,
        ReplyRequestMessage: router,           # Route to specialist agents
        TechnicalQuestion: tech_agent,
        BillingQuestion: billing_agent,
        GeneralQuestion: general_agent,
        AgentReply: orchestrator,             # Collect responses
    },
)

The orchestrator becomes a smart router to specialized agents.

# Process multiple messages concurrently
results = await asyncio.gather(
    bus.process(message1),
    bus.process(message2),
    bus.process(message3),
)

Each message flow is independent and can run concurrently.

# Nest systems arbitrarily deep
outer_system = CompositeProcessor(
    handlers={
        SystemInput: preprocessor,
        PreprocessedInput: inner_system,     # Nested
        SystemOutput: postprocessor,
    },
)

inner_system = CompositeProcessor(
    handlers={
        PreprocessedInput: sub_system,       # Double nested!
        SubOutput: formatter,
    },
)

Composites within composites within composites. Same pattern, any depth.

Real-World Production Patterns

Error handling and fallbacks

class RobustCRMProcessor(MessageProcessor):
    def __init__(self, primary_crm: CRMClient, fallback_crm: CRMClient):
        self._primary = primary_crm
        self._fallback = fallback_crm
    
    async def _process(self, message: IngressMessage) -> Optional[Message]:
        try:
            user_info = await self._primary.get_user(message.user_id)
        except CRMError:
            logger.warning("Primary CRM failed, trying fallback")
            try:
                user_info = await self._fallback.get_user(message.user_id)
            except CRMError:
                # Return default user info instead of failing
                user_info = {"name": "Valued Customer", "tier": "standard"}
        
        return CRMMessage(user_info=user_info, ...)

Processors handle their own errors. The framework doesn’t dictate policy.

Rate limiting

class RateLimitedProcessor(MessageProcessor):
    def __init__(self, inner: MessageProcessor, rate_limiter: RateLimiter):
        self._inner = inner
        self._rate_limiter = rate_limiter
    
    async def _process(self, message: Message) -> Optional[Message]:
        await self._rate_limiter.acquire(message.user_id)
        return await self._inner.process(message)

Wrap processors with rate limiting. Composition over framework features.

Circuit breaker

class CircuitBreakerProcessor(MessageProcessor):
    def __init__(self, inner: MessageProcessor, breaker: CircuitBreaker):
        self._inner = inner
        self._breaker = breaker
    
    async def _process(self, message: Message) -> Optional[Message]:
        if self._breaker.is_open():
            logger.warning("Circuit breaker open, skipping processor")
            return None
        
        try:
            result = await self._inner.process(message)
            self._breaker.record_success()
            return result
        except Exception as e:
            self._breaker.record_failure()
            raise

Standard patterns, implemented as wrappers.

Key Insights

Types Document Flow

Custom message types make the enrichment pipeline self-documenting. Read the types, understand the system.

Processors Are Independent

CRM, DB, Orchestrator, ReplyAgent—each testable in isolation, swappable, and reusable.

Composition Creates Structure

Nested composites organize complex systems into logical subsystems with clear boundaries.

Routing Is Automatic

Type-based routing eliminates manual chaining. Define handlers, let the composite wire them up.

What Makes This Production-Ready

Each stage can fail independently

If CRM fails, return None and stop. If DB fails, return None. If reply agent fails, orchestrator can return a default response. Failures are explicit, not cascading.

Easy to test each component

Mock the CRM response, test DB processor. Mock enriched messages, test agent logic. Unit test each processor, integration test the composite.

Observable end-to-end

Observers track every transformation, calculate latencies, log errors. Full visibility without code changes.

Scales horizontally

Run multiple instances of the bus, process messages concurrently. Stateless messages + stateful processors = horizontal scaling.

Try It Yourself

Run the example

Copy the full code and run it. Watch the logs to see:

CRM lookup for user info
Database fetch for history
Orchestrator routing to reply agent
Agent generating personalized response
Orchestrator approving and returning

Extend the enrichment

Add another processor between CRM and DB:

class PreferencesProcessor(MessageProcessor):
    async def _process(self, message: CRMMessage) -> PreferencesMessage:
        prefs = await self._load_preferences(message.user_id)
        return PreferencesMessage(
            user_info=message.user_info,
            preferences=prefs,
            user_message=message.user_message,
        )

Update the handlers, and it just works.

Add more agents

Create specialist agents for different question types:

class TechnicalAgent(MessageProcessor): ...
class BillingAgent(MessageProcessor): ...

# Update orchestrator to route based on message content
class SmartOrchestrator(MessageProcessor):
    async def _process(self, message: EnrichedMessage) -> Message:
        if "technical" in message.user_message.content.lower():
            return TechnicalQuestion(...)
        elif "billing" in message.user_message.content.lower():
            return BillingQuestion(...)
        else:
            return GeneralQuestion(...)

Add observability

Create a metrics observer:

class MetricsObserver(BaseObserver):
    async def on_message_processed(self, message: MessageProcessed):
        # Send to your metrics system
        metrics.increment(f"{message.source_processor.name}.messages")
        metrics.timing(f"{message.source_processor.name}.latency", latency)

Comparing Complexity

Without tinychat’s primitives:

# Manual orchestration, coupled logic
async def handle_message(user_message: str, user_id: str):
    user_info = await crm_client.get_user(user_id)
    history = await db.get_history(user_id)
    request = format_request(user_message, user_info, history)
    reply = await agent.generate(request)
    approved_reply = approve_reply(reply)
    return approved_reply

# Everything coupled, hard to test, hard to extend
# Where do you add caching? Retry logic? Metrics?
# How do you compose this with other flows?

With tinychat’s primitives:

# Declarative topology, automatic routing
bus = CompositeProcessor(
    handlers={
        IngressMessage: crm,
        CRMMessage: db,
        EnrichedMessage: agent_bus,
    },
)

result = await bus.process(message)

# Each processor independent, testable, reusable
# Add caching: wrap processor. Add retry: wrap processor.
# Add metrics: add observer. Compose: nest composites.

tinychat turns complex orchestration into simple composition. The primitives do the heavy lifting.

Ready for more?

Explore the information theory foundations and advanced topology patterns in the Core Primitives guide.

Primitives

Examples

​What We’re Building

​Message Types: Documenting the Flow

​Stage 1: CRM Enrichment

​Real-World Extensions

​Stage 2: Database Enrichment

​Automatic Chaining

​Stage 3: Agent System (Nested Composite)

​The Orchestrator: State-Dependent Routing (1:N)

​Flow Through the Agent Bus

​Why This Pattern?

Separation of Concerns

Easy to Extend

Testable in Isolation

Composable State Machines

​The Reply Agent: Context-Aware Generation

​Swapping Implementations

​Observability: Tracking the Journey

​Lifecycle: Assembly and Execution

​Topology Visualization

​Scaling Patterns

​Real-World Production Patterns

​Key Insights

Types Document Flow

Processors Are Independent

Composition Creates Structure

Routing Is Automatic

​What Makes This Production-Ready

​Try It Yourself

​Comparing Complexity

Ready for more?

What We’re Building

Message Types: Documenting the Flow

Stage 1: CRM Enrichment

Real-World Extensions

Stage 2: Database Enrichment

Automatic Chaining

Stage 3: Agent System (Nested Composite)

The Orchestrator: State-Dependent Routing (1:N)

Flow Through the Agent Bus

Why This Pattern?

The Reply Agent: Context-Aware Generation

Swapping Implementations

Observability: Tracking the Journey

Lifecycle: Assembly and Execution

Topology Visualization

Scaling Patterns

Real-World Production Patterns

Key Insights

What Makes This Production-Ready

Try It Yourself

Comparing Complexity