===============
LIBRARY RULES
===============
From library maintainers:
- Always use async functions for agents
- Use @clutch.agent() decorator to register agents
- Use Terminate(result) to stop pipeline and return
- Use Handover(agent_name, data) to transfer control
- Use Pydantic models for type-safe data flow
- Choose strategy based on use case: SEQUENTIAL for pipelines, SELECTOR for routing, GRAPH for workflows, ROUND_ROBIN for iteration

### Complete RAG Pipeline Example in Python

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

Provides a full example of a Retrieval Augmented Generation (RAG) pipeline using EggAI Clutch. It includes agents for loading documents, chunking, embedding, summarizing, and indexing. This example requires pydantic and hashlib.

```python
import asyncio
import hashlib
from pydantic import BaseModel
from eggai_clutch import Clutch, Terminate

class Document(BaseModel):
    path: str
    content: str = ""
    doc_id: str = ""
    chunks: list[str] = []
    embeddings: list[list[float]] = []
    summary: str = ""

clutch = Clutch("rag-indexer")

@clutch.agent()
async def loader(doc: Document) -> Document:
    # Simulate loading file content
    doc.content = f"Content from {doc.path}" * 100
    doc.doc_id = hashlib.md5(doc.path.encode()).hexdigest()[:8]
    return doc

@clutch.agent()
async def chunker(doc: Document) -> Document:
    chunk_size = 200
    doc.chunks = [
        doc.content[i:i+chunk_size]
        for i in range(0, len(doc.content), chunk_size)
    ]
    return doc

@clutch.agent()
async def embedder(doc: Document) -> Document:
    # Simulate embedding API call
    doc.embeddings = [[0.1, 0.2, 0.3] for _ in doc.chunks]
    return doc

@clutch.agent()
async def summarizer(doc: Document) -> Document:
    # Simulate LLM API call
    doc.summary = f"Summary of document with {len(doc.chunks)} chunks"
    return doc

@clutch.agent()
async def indexer(doc: Document) -> Document:
    # Store in vector database
    print(f"Indexed {doc.doc_id} with {len(doc.embeddings)} embeddings")
    raise Terminate(doc)

# result = await clutch.run(Document(path="/docs/report.pdf"))
# print(f"Processed: {result['doc_id']}")
# print(f"Chunks: {len(result['chunks'])}")
# print(f"Summary: {result['summary']}")
```

--------------------------------

### Sequential Strategy Example

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Illustrates the default SEQUENTIAL strategy where agents execute in the order they are defined. Data flows from one agent to the next until termination.

```python
from eggai_clutch import Clutch, Terminate

clutch = Clutch("pipeline")

@clutch.agent()
async def step_a(data): ...

@clutch.agent()
async def step_b(data): ...

@clutch.agent()
async def step_c(data):
    raise Terminate(data)
```

--------------------------------

### Quick Start: Define and Run a Pipeline

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Demonstrates a basic pipeline with two agents, 'chunker' and 'summarizer', using pydantic models for data. The 'chunker' divides content into chunks, and 'summarizer' creates a summary, terminating the pipeline.

```python
import asyncio
from pydantic import BaseModel
from eggai_clutch import Clutch, Terminate

class Document(BaseModel):
    content: str
    chunks: list[str] = []
    summary: str = ""

clutch = Clutch("rag-pipeline")

@clutch.agent()
async def chunker(doc: Document) -> Document:
    doc.chunks = [doc.content[i:i+500] for i in range(0, len(doc.content), 500)]
    return doc

@clutch.agent()
async def summarizer(doc: Document) -> Document:
    doc.summary = f"Summary of {len(doc.chunks)} chunks"
    raise Terminate(doc)

async def main():
    result = await clutch.run(Document(content="..." * 1000))
    print(result["summary"])

asyncio.run(main())
```

--------------------------------

### Task Submission and Streaming in Python

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

Illustrates how to submit tasks to a Clutch instance, retrieve results with timeouts, and stream step events. It also shows how to cancel a running task. This example requires pydantic.

```python
from eggai_clutch import Clutch, StepEvent
from pydantic import BaseModel

class Job(BaseModel):
    id: int
    status: str = "pending"

clutch = Clutch("jobs")

@clutch.agent()
async def validate(job: Job) -> Job:
    job.status = "validated"
    return job

@clutch.agent()
async def process(job: Job) -> Job:
    job.status = "processed"
    return job

@clutch.agent()
async def complete(job: Job) -> Job:
    job.status = "complete"
    raise Terminate(job)

# Submit and get handle immediately
# task = await clutch.submit(Job(id=1))
# print(f"Task ID: {task.id}")
# print(f"Done: {task.done}")  # False

# Wait for result with timeout
# result = await task.result(timeout=10.0)

# Or await directly
# result = await task

# Stream step events
# async for event in clutch.stream(Job(id=2)):
#     print(f"Step: {event.step}, Data: {event.data}, Final: {event.final}")

# Cancel a task
# task.cancel()
```

--------------------------------

### Selector Strategy Example for Dynamic Routing

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Demonstrates the SELECTOR strategy, where a dedicated 'route' function dynamically determines the next agent to execute based on input conditions. This is useful for intent classification.

```python
from eggai_clutch import Clutch, Strategy, Terminate

clutch = Clutch("router", Strategy.SELECTOR)

@clutch.selector
async def route(ticket: Ticket) -> str:
    if "billing" in ticket.query:
        return "billing"
    return "general"

@clutch.agent()
async def billing(ticket: Ticket):
    ticket.response = "Billing handled"
    raise Terminate(ticket)

@clutch.agent()
async def general(ticket: Ticket):
    ticket.response = "General handled"
    raise Terminate(ticket)
```

--------------------------------

### Install eggai-clutch

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Installs the eggai-clutch Python package using pip. This is the primary method for adding the library to your project.

```bash
pip install eggai-clutch
```

--------------------------------

### Graph Strategy Example for Workflow Definition

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Shows the GRAPH strategy, allowing explicit definition of edges between agents to create complex, non-linear workflows. Agents are linked by their names as strings.

```python
from eggai_clutch import Clutch, Strategy

clutch = Clutch("workflow", Strategy.GRAPH)

@clutch.agent(edges=["validate"])
async def parse(data): ...

@clutch.agent(edges=["store"])
async def validate(data): ...

@clutch.agent()  # No edges = terminal
async def store(data):
    raise Terminate(data)
```

--------------------------------

### Dynamic Routing with Selector Strategy in Python

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

Demonstrates dynamic routing of support tickets to different agents using a selector strategy. It defines agents for billing, technical, and general inquiries and uses a selector function to determine the appropriate agent based on the query. This example requires pydantic.

```python
from eggai_clutch import Clutch, Strategy, Terminate
from pydantic import BaseModel

class SupportTicket(BaseModel):
    query: str
    category: str = ""
    response: str = ""

clutch = Clutch("support-router", Strategy.SELECTOR, max_turns=5)

@clutch.selector()
async def route(ticket: SupportTicket) -> str:
    """Returns the name of the agent to route to"""
    query = ticket.query.lower()
    if "billing" in query or "payment" in query:
        return "billing"
    elif "bug" in query or "error" in query:
        return "technical"
    else:
        return "general"

@clutch.agent()
async def billing(ticket: SupportTicket) -> SupportTicket:
    ticket.category = "billing"
    ticket.response = "Billing team will assist you"
    raise Terminate(ticket)

@clutch.agent()
async def technical(ticket: SupportTicket) -> SupportTicket:
    ticket.category = "technical"
    ticket.response = "Technical support initiated"
    raise Terminate(ticket)

@clutch.agent()
async def general(ticket: SupportTicket) -> SupportTicket:
    ticket.category = "general"
    ticket.response = "General inquiry handled"
    raise Terminate(ticket)

# Test routing
# result = await clutch.run(SupportTicket(query="I was charged twice"))
# print(f"Category: {result['category']}")  # billing
# print(f"Response: {result['response']}")
```

--------------------------------

### Distributed Execution with Redis Transport in Python

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

Shows how to perform distributed execution using EggAI Clutch with a Redis transport. It initializes Clutch with a Redis transport and defines agents that can run on different workers. This example requires pydantic and redis.

```python
from eggai import RedisTransport
from eggai_clutch import Clutch, Terminate
from pydantic import BaseModel

class Data(BaseModel):
    value: int
    squared: int = 0

# Initialize with transport for distributed execution
transport = RedisTransport(url="redis://localhost:6379")
clutch = Clutch("distributed-pipeline", transport=transport)

@clutch.agent()
async def square(data: Data) -> Data:
    data.squared = data.value ** 2
    raise Terminate(data)

# Submit work - automatically routes to distributed workers
# task = await clutch.submit(Data(value=5))
# result = await task.result(timeout=30.0)
# print(f"Result: {result['squared']}")  # 25

# Cleanup resources
# await clutch.stop()

# Or use context manager
# async with Clutch("pipeline", transport=transport) as clutch:
#     @clutch.agent()
#     async def handler(data: Data) -> Data:
#         raise Terminate(data)
# 
#     result = await clutch.run(Data(value=10))
```

--------------------------------

### Register Agents with Decorators and Define Graph Edges in EggAI Clutch

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

This example illustrates registering agents using the `@clutch.agent()` decorator in EggAI Clutch. It covers simple agent registration and how to define explicit edges for the GRAPH strategy, specifying the flow between agents using the `edges` parameter. Terminal nodes are agents without specified outgoing edges.

```python
from pydantic import BaseModel
from eggai_clutch import Clutch

class Task(BaseModel):
    data: str
    processed: bool = False

clutch = Clutch("example")

# Simple agent registration
@clutch.agent()
async def process(task: Task) -> Task:
    task.data = task.data.upper()
    task.processed = True
    return task

# Graph strategy with explicit edges
graph_clutch = Clutch("graph-example", Strategy.GRAPH)

@graph_clutch.agent(edges=["validate"])
async def parse(task: Task) -> Task:
    return task

@graph_clutch.agent(edges=["store"])
async def validate(task: Task) -> Task:
    return task

@graph_clutch.agent()  # No edges means terminal node
async def store(task: Task) -> Task:
    raise Terminate(task)
```

--------------------------------

### Initialize EggAI Clutch with Different Strategies and Hooks

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

This snippet shows various ways to initialize the EggAI Clutch framework with different execution strategies: SEQUENTIAL, SELECTOR, GRAPH, and ROUND_ROBIN. It also demonstrates how to configure monitoring hooks (`on_request`, `on_response`, `on_step`) for observing pipeline execution.

```python
from eggai_clutch import Clutch, Strategy

# Sequential strategy (default) - agents run in order
sequential = Clutch("pipeline", Strategy.SEQUENTIAL, max_turns=100)

# Selector strategy - dynamic routing based on selector function
selector = Clutch("router", Strategy.SELECTOR, max_turns=50)

# Graph strategy - explicit edges between agents
graph = Clutch("workflow", Strategy.GRAPH)

# Round-robin strategy - cycle through agents
round_robin = Clutch("iterative", Strategy.ROUND_ROBIN, max_turns=10)

# With hooks for monitoring
monitored = Clutch(
    "tracked-pipeline",
    on_request=lambda req: print(f"Starting: {req}"),
    on_response=lambda req, res: print(f"Completed: {res}"),
    on_step=lambda step, inp, out: print(f"Step {step}: {inp} -> {out}")
)
```

--------------------------------

### Create and Run a Sequential Agent Pipeline with EggAI Clutch

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

This snippet demonstrates how to create a sequential agent pipeline using EggAI Clutch. It defines Pydantic models for data, registers agents using decorators, and runs the pipeline to process data. The `Terminate` exception is used to end the pipeline execution and return a result.

```python
import asyncio
from pydantic import BaseModel
from eggai_clutch import Clutch, Terminate

class Document(BaseModel):
    content: str
    chunks: list[str] = []
    summary: str = ""

clutch = Clutch("rag-pipeline")

@clutch.agent()
async def chunker(doc: Document) -> Document:
    doc.chunks = [doc.content[i:i+500] for i in range(0, len(doc.content), 500)]
    return doc

@clutch.agent()
async def summarizer(doc: Document) -> Document:
    doc.summary = f"Summary of {len(doc.chunks)} chunks"
    raise Terminate(doc)

async def main():
    result = await clutch.run(Document(content="..." * 1000))
    print(f"Processed {len(result['chunks'])} chunks")
    print(f"Summary: {result['summary']}")

asyncio.run(main())
```

--------------------------------

### Configuring Hooks for Request, Response, and Step Events

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Demonstrates how to configure callback functions for specific events during clutch execution: `on_request` (before processing), `on_response` (after completion), and `on_step` (after each agent step).

```python
clutch = Clutch(
    "pipeline",
    on_request=async_fn,   # Before processing
    on_response=async_fn,  # After completion
    on_step=async_fn,      # After each step
)
```

--------------------------------

### Task API: Running and Managing Tasks

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Shows how to interact with the Clutch task API, including submitting tasks, waiting for results, checking task status, and cancelling tasks. It also covers streaming step events.

```python
# Submit and wait
result = await clutch.run(input_data, timeout=30.0)

# Submit and get handle
task = await clutch.submit(input_data)
print(task.done)        # Non-blocking check
result = await task     # Await directly
result = await task.result(timeout=10.0)
task.cancel()

# Stream step events
async for event in clutch.stream(input_data):
    print(f"Step: {event.step}, Final: {event.final}")
```

--------------------------------

### Control Flow: Handover to Another Agent

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Illustrates the `Handover` exception, which transfers execution control to a specified agent. This is useful for routing tasks to specialized agents within a pipeline.

```python
from eggai_clutch import Handover

@clutch.agent()
async def router(data):
    if needs_review:
        raise Handover("reviewer", data)
    return data
```

--------------------------------

### Control Agent Pipeline Flow with Terminate and Handover in EggAI Clutch

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

This snippet demonstrates advanced control flow mechanisms in EggAI Clutch using `Terminate` and `Handover` exceptions. `Terminate` stops the pipeline and returns a result, while `Handover` transfers execution to a specific agent with potentially modified data, enabling complex branching and early exits.

```python
from eggai_clutch import Clutch, Terminate, Handover
from pydantic import BaseModel

class Request(BaseModel):
    priority: str
    data: dict

clutch = Clutch("control-flow")

@clutch.agent()
async def router(req: Request) -> Request:
    # Stop pipeline immediately and return result
    if req.priority == "critical":
        raise Terminate({"status": "escalated", "data": req.data})

    # Transfer to specific agent with modified data
    if req.priority == "high":
        req.data["urgent"] = True
        raise Handover("priority_handler", req)

    return req

@clutch.agent()
async def priority_handler(req: Request) -> Request:
    req.data["processed"] = True
    raise Terminate(req)

@clutch.agent()
async def normal_handler(req: Request) -> Request:
    raise Terminate(req)

# Run the pipeline
result = await clutch.run(Request(priority="high", data={"id": 123}))
```

--------------------------------

### Python Graph Workflow for Order Processing with Conditional Logic

Source: https://context7.com/eggai-tech/eggai-clutch/llms.txt

This Python code defines an order processing workflow using EggAI Clutch's GRAPH strategy. It includes agents for validation, fraud checking, processing, and notification, with conditional logic to skip the fraud check for orders below a certain total. It uses Pydantic for data modeling and exception-based control flow for agent transitions.

```python
from eggai_clutch import Clutch, Strategy, Terminate, Handover
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    total: float
    validated: bool = False
    fraud_checked: bool = False
    processed: bool = False

graph = Clutch("order-workflow", Strategy.GRAPH)

@graph.agent(edges=["fraud_check"])
async def validate(order: Order) -> Order:
    order.validated = True
    if order.total > 10000:
        # Skip fraud check for small orders
        raise Handover("process", order)
    return order

@graph.agent(edges=["process"])
async def fraud_check(order: Order) -> Order:
    order.fraud_checked = True
    return order

@graph.agent(edges=["notify"])
async def process(order: Order) -> Order:
    order.processed = True
    return order

@graph.agent()
async def notify(order: Order) -> Order:
    print(f"Order {order.id} completed")
    raise Terminate(order)

result = await graph.run(Order(id=123, total=5000))
print(f"Validated: {result['validated']}")
print(f"Fraud checked: {result['fraud_checked']}")
print(f"Processed: {result['processed']}")
```

--------------------------------

### Control Flow: Terminate an Agent

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Demonstrates how to use `Terminate` to stop agent execution and return a specific result. This is used for completing a pipeline or a branch of execution.

```python
from eggai_clutch import Terminate

@clutch.agent()
async def final_step(data):
    raise Terminate({"status": "done", "result": data})
```

--------------------------------

### Distributed Mode: Using Redis Transport

Source: https://github.com/eggai-tech/eggai-clutch/blob/main/README.md

Configures Clutch for distributed execution by adding a transport, specifically RedisTransport. This allows agents to run across multiple processes or machines.

```python
from eggai import RedisTransport

transport = RedisTransport(url="redis://localhost:6379")
clutch = Clutch("pipeline", transport=transport)

# Worker process
await clutch.submit(data)  # Routes to distributed workers

# Cleanup
await clutch.stop()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.