Event-Driven AI Pattern

import asyncio
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Callable, Optional
from collections import defaultdict


class JobStatus(Enum):
    QUEUED = "queued"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Event:
    type: str
    payload: dict
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Job:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    event: Event = None
    status: JobStatus = JobStatus.QUEUED
    result: Optional[dict] = None
    error: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None


class EventBus:
    """Simple in-process event bus for event-driven AI."""

    def __init__(self):
        self.handlers: dict[str, list[Callable]] = defaultdict(list)
        self.event_log: list[Event] = []

    def subscribe(self, event_type: str, handler: Callable):
        """Register a handler for an event type."""
        self.handlers[event_type].append(handler)

    async def publish(self, event: Event):
        """Publish an event to all subscribers."""
        self.event_log.append(event)
        handlers = self.handlers.get(event.type, [])
        tasks = [handler(event) for handler in handlers]
        if tasks:
            await asyncio.gather(*tasks,
                                 return_exceptions=True)


class AsyncJobQueue:
    """Async job queue for AI processing tasks."""

    def __init__(self, max_workers: int = 3):
        self.queue: asyncio.Queue = asyncio.Queue()
        self.jobs: dict[str, Job] = {}
        self.max_workers = max_workers
        self.event_bus = EventBus()
        self._workers_started = False

    async def submit(self, event: Event) -> str:
        """Submit an event for async processing."""
        job = Job(event=event)
        self.jobs[job.id] = job
        await self.queue.put(job)

        # Publish job queued event
        await self.event_bus.publish(Event(
            type="job.queued",
            payload={"job_id": job.id,
                     "event_type": event.type}
        ))
        return job.id

    def get_status(self, job_id: str) -> Optional[dict]:
        """Check the status of a submitted job."""
        job = self.jobs.get(job_id)
        if not job:
            return None
        return {
            "job_id": job.id,
            "status": job.status.value,
            "result": job.result,
            "error": job.error,
            "created_at": job.created_at.isoformat(),
            "completed_at": (job.completed_at.isoformat()
                            if job.completed_at else None)
        }

    async def _worker(self, worker_id: int,
                      process_fn: Callable):
        """Worker that processes jobs from the queue."""
        while True:
            job = await self.queue.get()
            job.status = JobStatus.PROCESSING

            await self.event_bus.publish(Event(
                type="job.processing",
                payload={"job_id": job.id,
                         "worker": worker_id}
            ))

            try:
                result = await process_fn(job.event)
                job.status = JobStatus.COMPLETED
                job.result = result
                job.completed_at = datetime.now()

                await self.event_bus.publish(Event(
                    type="job.completed",
                    payload={"job_id": job.id,
                             "result": result}
                ))
            except Exception as e:
                job.status = JobStatus.FAILED
                job.error = str(e)
                job.completed_at = datetime.now()

                await self.event_bus.publish(Event(
                    type="job.failed",
                    payload={"job_id": job.id,
                             "error": str(e)}
                ))
            finally:
                self.queue.task_done()

    async def start_workers(self, process_fn: Callable):
        """Start worker tasks to process the queue."""
        if self._workers_started:
            return
        self._workers_started = True
        for i in range(self.max_workers):
            asyncio.create_task(
                self._worker(i, process_fn)
            )

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import anthropic
import json

app = FastAPI()


async def stream_claude_response(prompt: str):
    """Stream Claude's response token by token via SSE."""
    client = anthropic.AsyncAnthropic()

    # SSE format: each event is "data: ...\n\n"
    async with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    ) as stream:
        async for text in stream.text_stream:
            # Format as Server-Sent Event
            event_data = json.dumps({"text": text})
            yield f"data: {event_data}\n\n"

    # Send completion event
    yield f"data: {json.dumps({'done': True})}\n\n"


@app.post("/api/chat/stream")
async def chat_stream(request: Request):
    """Endpoint that streams AI responses via SSE."""
    body = await request.json()
    prompt = body.get("prompt", "")

    return StreamingResponse(
        stream_claude_response(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )


# Client-side JavaScript to consume the stream
CLIENT_JS = """
async function streamChat(prompt) {
    const response = await fetch('/api/chat/stream', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
    });

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\\n\\n');
        buffer = lines.pop();  // Keep incomplete line

        for (const line of lines) {
            if (line.startsWith('data: ')) {
                const data = JSON.parse(line.slice(6));
                if (data.done) {
                    console.log('Stream complete');
                } else {
                    // Append token to the UI
                    document.getElementById('output')
                        .textContent += data.text;
                }
            }
        }
    }
}
"""