agentle 0.9.47
...
Requires Python: >=3.13

<div align="center">
  <img src="/docs/logo.png" alt="Agentle Logo" width="200"/>
  
  <h3>✨ <em>Elegantly Simple AI Agents for Production</em> ✨</h3>
  
  <p>
    <strong>Build powerful AI agents with minimal code, maximum control</strong>
  </p>

  <p>
    <a href="https://www.python.org/downloads/"><img src="https://img.shields.io/badge/python-3.13+-blue.svg" alt="Python 3.13+"></a>
    <a href="https://opensource.org/licenses/MIT"><img src="https://img.shields.io/badge/license-MIT-green.svg" alt="MIT License"></a>
    <a href="https://badge.fury.io/py/agentle"><img src="https://badge.fury.io/py/agentle.svg" alt="PyPI version"></a>
  </p>

  <p>
    <a href="#-quick-start">Quick Start</a> •
    <a href="#-why-agentle">Why Agentle</a> •
    <a href="#-features">Features</a> •
    <a href="#-showcase">Showcase</a> •
    <a href="#-documentation">Docs</a>
  </p>
</div>

---

## 🎯 Why Agentle?

<table>
<tr>
<td width="50%">

### 🚀 **Simple Yet Powerful**
```python
# Just 5 lines to create an AI agent
agent = Agent(
    name="Assistant",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You are a helpful assistant."
)

response = agent.run("Agentle is cool.")
```

</td>
<td width="50%">

### 🏗️ **Production Ready**
- 🔍 **Built-in Observability** with Langfuse
- 🌐 **Instant APIs** with automatic documentation
- 💪 **Type-Safe** with full type hints
- 🎯 **Structured Outputs** with Pydantic
- 🔧 **Tool Calling** support out of the box
- 📦 **Minimum dependencies** you only install what you need

</td>
</tr>
</table>

## ⚡ Quick Start

### Installation

```bash
pip install agentle
```

### Your First Agent in 30 Seconds

```python
from agentle.agents.agent import Agent
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create a simple agent
agent = Agent(
    name="Quick Start Agent",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You are a helpful assistant who provides concise information."
)

# Run the agent
response = agent.run("What are the three laws of robotics?")
print(response.text)
```

## 🌟 Features

<div align="center">

| 🎨 **Beautiful UIs** | 🌐 **Instant APIs** | 📊 **Observability** | 🗄️ **Production Caching** |
|:---:|:---:|:---:|:---:|
| Create chat interfaces with Streamlit in minutes | Deploy RESTful APIs with automatic Scalar docs | Track everything with built-in Langfuse integration | Flexible caching with InMemory & Redis stores |
| ![Streamlit UI](https://github.com/user-attachments/assets/1c31da4c-aeb2-4ca6-88ac-62fb903d6d92) | ![API Docs](https://github.com/user-attachments/assets/d9d743cb-ad9c-41eb-a059-eda089efa6b6) | ![Tracing](https://github.com/user-attachments/assets/94937238-405c-4011-83e2-147cec5cf3e7) | Intelligent document caching for performance |

</div>

### 🔥 Core Capabilities

<details>
<summary><b>🗄️ Production-Ready Caching</b> - Intelligent document caching for performance and scalability</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.knowledge.static_knowledge import StaticKnowledge
from agentle.parsing.cache import InMemoryDocumentCacheStore, RedisCacheStore
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# In-Memory Cache (Development & Single Process)
memory_cache = InMemoryDocumentCacheStore(cleanup_interval=300)

agent = Agent(
    name="Research Assistant",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You analyze documents efficiently with caching.",
    static_knowledge=[
        StaticKnowledge(content="large_report.pdf", cache=3600),  # Cache for 1 hour
        StaticKnowledge(content="https://api.example.com/data", cache="infinite"),  # Cache indefinitely
    ],
    document_cache_store=memory_cache
)

# Redis Cache (Production & Distributed)
redis_cache = RedisCacheStore(
    redis_url="redis://localhost:6379/0",
    key_prefix="agentle:docs:",
    default_ttl=7200  # 2 hours default
)

production_agent = Agent(
    name="Production Assistant",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You handle high-volume document processing.",
    static_knowledge=[
        StaticKnowledge(content="enterprise_docs.pdf", cache=86400),  # Cache for 1 day
    ],
    document_cache_store=redis_cache
)

# Cache Management
stats = memory_cache.get_stats()
print(f"Active cache entries: {stats['active_entries']}")

# Clear cache when needed
await memory_cache.clear_async()
```

**Benefits:**
- 🚀 **Performance**: Avoid re-parsing large documents
- 💰 **Cost Efficiency**: Reduce API calls for URL-based documents
- 📊 **Scalability**: Share cached documents across processes (Redis)
- 🔄 **Consistency**: Same parsed content across multiple runs
</details>

<details>
<summary><b>🤖 Intelligent Agents</b> - Build specialized agents with knowledge, tools, and structured outputs</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.knowledge.static_knowledge import StaticKnowledge
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider
from pydantic import BaseModel
from typing import List

# Define structured output
class WeatherForecast(BaseModel):
    location: str
    current_temperature: float
    conditions: str
    forecast: List[str]

# Create a weather tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    weather_data = {
        "New York": "Sunny, 75°F",
        "London": "Rainy, 60°F",
        "Tokyo": "Cloudy, 65°F",
    }
    return weather_data.get(location, f"Weather data not available for {location}")

# Build the agent
weather_agent = Agent(
    name="Weather Assistant",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You are a weather forecasting assistant.",
    # Add domain knowledge
    static_knowledge=[
        StaticKnowledge(content="weather_data/climate_patterns.pdf", cache=3600),
        "A heat wave is defined as a period of abnormally hot weather generally lasting more than two days."
    ],
    # Add tools
    tools=[get_weather],
    # Ensure structured responses
    response_schema=WeatherForecast
)

# Get typed responses
response = weather_agent.run("What's the weather like in Tokyo?")
forecast = response.parsed
print(f"Weather in {forecast.location}: {forecast.current_temperature}°C, {forecast.conditions}")
```
</details>

<details>
<summary><b>🔗 Agent Pipelines</b> - Chain agents for complex workflows</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.agent_pipeline import AgentPipeline
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create provider for reuse
provider = GoogleGenerationProvider()

# Create specialized agents
research_agent = Agent(
    name="Research Agent",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="""You are a research agent focused on gathering information.
    Be thorough and prioritize accuracy over speculation."""
)

analysis_agent = Agent(
    name="Analysis Agent",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="""You are an analysis agent that identifies patterns.
    Highlight meaningful relationships and insights from the data."""
)

summary_agent = Agent(
    name="Summary Agent",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="""You are a summary agent that creates concise summaries.
    Present key findings in a logical order with accessible language."""
)

# Create a pipeline
pipeline = AgentPipeline(
    agents=[research_agent, analysis_agent, summary_agent],
    debug_mode=True  # Enable to see intermediate steps
)

# Run the pipeline
result = pipeline.run("Research the impact of artificial intelligence on healthcare")
print(result.text)
```
</details>

<details>
<summary><b>👥 Agent Teams</b> - Dynamic orchestration with intelligent routing</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.agent_team import AgentTeam
from agentle.agents.a2a.models.agent_skill import AgentSkill
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create provider for reuse
provider = GoogleGenerationProvider()

# Create specialized agents with different skills
research_agent = Agent(
    name="Research Agent",
    description="Specialized in finding accurate information on various topics",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You are a research agent focused on gathering accurate information.",
    skills=[
        AgentSkill(name="search", description="Find information on any topic"),
        AgentSkill(name="fact-check", description="Verify factual claims"),
    ],
)

coding_agent = Agent(
    name="Coding Agent",
    description="Specialized in writing and debugging code",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You are a coding expert focused on writing clean, efficient code.",
    skills=[
        AgentSkill(name="code-generation", description="Write code in various languages"),
        AgentSkill(name="debugging", description="Find and fix bugs in code"),
    ],
)

# Create a team with these agents
team = AgentTeam(
    agents=[research_agent, coding_agent],
    orchestrator_provider=provider,
    orchestrator_model="gemini-2.5-flash",
)

# Run the team with different queries
research_query = "What are the main challenges in quantum computing today?"
research_result = team.run(research_query)

coding_query = "Write a Python function to find the Fibonacci sequence up to n terms."
coding_result = team.run(coding_query)
```
</details>

<details>
<summary><b>🔌 MCP Integration</b> - Connect to external tools via Model Context Protocol</summary>

```python
from agentle.agents.agent import Agent
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider
from agentle.mcp.servers.streamable_http_mcp_server import StreamableHTTPMCPServer
from agentle.mcp.servers.stdio_mcp_server import StdioMCPServer
from agentle.mcp.session_management import RedisSessionManager

# Set up provider
provider = GoogleGenerationProvider()

# Create MCP servers
stdio_server = StdioMCPServer(
    server_name="File System MCP",
    command="/path/to/filesystem_mcp_server",  # Replace with actual command
    server_env={"DEBUG": "1"},
)

# For development (single-process environments)
sse_server_dev = StreamableHTTPMCPServer(
    server_name="Weather API MCP",
    server_url="http://localhost:3000",  # Replace with actual server URL
)

# For production (multi-process environments)
redis_session = RedisSessionManager(
    redis_url="redis://redis-server:6379/0",
    key_prefix="agentle_mcp:",
    expiration_seconds=3600  # 1 hour session lifetime
)

sse_server_prod = StreamableHTTPMCPServer(
    server_name="Weather API MCP",
    server_url="https://api.example.com",
    session_manager=redis_session
)

# Create agent with MCP servers
agent = Agent(
    name="MCP-Augmented Assistant",
    description="An assistant that can access external tools via MCP",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You are a helpful assistant with access to external tools.",
    mcp_servers=[stdio_server, sse_server_dev],
)

# Use the start_mcp_servers context manager for proper connection handling
with agent.start_mcp_servers():
    # Query that uses MCP server tools
    response = agent.run("What's the weather like in Tokyo today?")
    print(response.generation.text)
```
</details>

## 🖼️ Visual Showcase

### 🎨 Build Beautiful Chat UIs

Transform your agent into a professional chat interface with just a few lines:

```python
from agentle.agents.agent import Agent
from agentle.agents.ui.streamlit import AgentToStreamlit
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create your agent
travel_agent = Agent(
    name="Travel Guide",
    description="A helpful travel guide that answers questions about destinations.",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="""You are a knowledgeable travel guide who helps users plan trips.""",
)

# Convert the agent to a Streamlit app
streamlit_app = AgentToStreamlit(
    title="Travel Assistant",
    description="Ask me anything about travel destinations and planning!",
    initial_mode="presentation",  # Can be "dev" or "presentation"
).adapt(travel_agent)

# Run the Streamlit app
if __name__ == "__main__":
    streamlit_app()
```

<img width="100%" alt="Streamlit Chat Interface" src="https://github.com/user-attachments/assets/1c31da4c-aeb2-4ca6-88ac-62fb903d6d92" />

### 🌐 Deploy Production APIs

Expose your agents, teams, and pipelines as RESTful APIs with automatic documentation:

<details>
<summary><b>🤖 Single Agent API</b> - Deploy individual agents as REST services</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.asgi.blacksheep.agent_to_blacksheep_application_adapter import AgentToBlackSheepApplicationAdapter
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create your agent
code_assistant = Agent(
    name="Code Assistant",
    description="An AI assistant specialized in helping with programming tasks.",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="""You are a helpful programming assistant.
    You can answer questions about programming languages, help debug code,
    explain programming concepts, and provide code examples.""",
)

# Convert the agent to a BlackSheep ASGI application
app = AgentToBlackSheepApplicationAdapter().adapt(code_assistant)

# Run the API server
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
```

**Generated API Endpoints:**
- `POST /api/v1/agents/code_assistant/run` - Execute the agent
- `POST /api/v1/agents/code_assistant/run/resume` - Resume suspended executions (HITL)
- `GET /docs` - Interactive API documentation
- `GET /openapi` - OpenAPI specification

</details>

<details>
<summary><b>👥 Agent Team API</b> - Deploy dynamic teams with intelligent orchestration</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.agent_team import AgentTeam
from agentle.agents.asgi.blacksheep.agent_to_blacksheep_application_adapter import AgentToBlackSheepApplicationAdapter
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

provider = GoogleGenerationProvider()

# Create specialized agents
research_agent = Agent(
    name="Research Agent",
    description="Specialized in finding and analyzing information",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You are a research expert focused on gathering accurate information.",
)

coding_agent = Agent(
    name="Coding Agent", 
    description="Specialized in writing and debugging code",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You are a coding expert that writes clean, efficient code.",
)

writing_agent = Agent(
    name="Writing Agent",
    description="Specialized in creating clear and engaging content",
    generation_provider=provider,
    model="gemini-2.5-flash", 
    instructions="You are a writing expert that creates compelling content.",
)

# Create a team with intelligent orchestration
team = AgentTeam(
    agents=[research_agent, coding_agent, writing_agent],
    orchestrator_provider=provider,
    orchestrator_model="gemini-2.5-flash"
)

# Deploy the team as an API
app = AgentToBlackSheepApplicationAdapter().adapt(team)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
```

**Generated API Endpoints:**
- `POST /api/v1/team/run` - Execute task with dynamic agent selection
- `POST /api/v1/team/resume` - Resume suspended team executions
- `GET /docs` - Interactive documentation with team composition details

</details>

<details>
<summary><b>🔗 Agent Pipeline API</b> - Deploy sequential processing workflows</summary>

```python
from agentle.agents.agent import Agent
from agentle.agents.agent_pipeline import AgentPipeline
from agentle.agents.asgi.blacksheep.agent_to_blacksheep_application_adapter import AgentToBlackSheepApplicationAdapter
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

provider = GoogleGenerationProvider()

# Create pipeline stages
data_processor = Agent(
    name="Data Processor",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You process and clean raw data, handling missing values and formatting.",
)

analyzer = Agent(
    name="Data Analyzer", 
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You perform statistical analysis and identify patterns in processed data.",
)

reporter = Agent(
    name="Report Generator",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You create comprehensive reports with insights and visualizations.",
)

# Create a sequential pipeline
pipeline = AgentPipeline(
    agents=[data_processor, analyzer, reporter],
    debug_mode=True  # Enable detailed logging
)

# Deploy the pipeline as an API
app = AgentToBlackSheepApplicationAdapter().adapt(pipeline)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)
```

**Generated API Endpoints:**
- `POST /api/v1/pipeline/run` - Execute sequential pipeline processing
- `POST /api/v1/pipeline/resume` - Resume suspended pipeline executions
- `GET /docs` - Interactive documentation with pipeline stage details

</details>

<details>
<summary><b>🔄 HITL-Enabled APIs</b> - Production APIs with human approval workflows</summary>

All API types (Agent, Team, Pipeline) automatically support Human-in-the-Loop workflows:

```python
from agentle.agents.errors.tool_suspension_error import ToolSuspensionError

def sensitive_operation(amount: float, account: str) -> str:
    """A tool that requires human approval for large amounts."""
    if amount > 10000:
        raise ToolSuspensionError(
            reason=f"Transfer of ${amount} requires approval",
            approval_data={"amount": amount, "account": account},
            timeout_seconds=3600  # 1 hour timeout
        )
    return f"Transfer completed: ${amount} to {account}"

# Any agent with HITL tools will automatically expose resume endpoints
agent_with_hitl = Agent(
    name="Financial Agent",
    tools=[sensitive_operation],
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You handle financial operations with human oversight.",
)

app = AgentToBlackSheepApplicationAdapter().adapt(agent_with_hitl)
```

**HITL API Flow:**
1. `POST /api/v1/agents/financial_agent/run` - Returns 202 with `resumption_token` if suspended
2. Human approves via external system (web UI, mobile app, etc.)
3. `POST /api/v1/agents/financial_agent/run/resume` - Continues execution with approval data

</details>

**🎯 Key API Features:**
- **📚 Automatic Documentation**: Interactive Scalar UI with detailed endpoint descriptions
- **🔄 HITL Support**: Built-in suspension/resumption for human approval workflows  
- **📊 Structured Responses**: Consistent `AgentRunOutput` format across all endpoints
- **⚡ Async Processing**: Non-blocking execution with proper error handling
- **🔍 Type Safety**: Full OpenAPI schema generation with request/response validation
- **🏗️ Production Ready**: Built on BlackSheep ASGI for high-performance deployment

<img width="100%" alt="API Documentation" src="https://github.com/user-attachments/assets/d9d743cb-ad9c-41eb-a059-eda089efa6b6" />

### 📊 Enterprise-Grade Observability

Monitor every aspect of your agents in production:

```python
from agentle.generations.tracing.langfuse_otel_client import LangfuseOtelClient
from agentle.agents.agent import Agent
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create a tracing client
tracing_client = LangfuseOtelClient()

# Create an agent with tracing enabled
agent = Agent(
    name="Traceable Agent",
    generation_provider=GoogleGenerationProvider(otel_clients=tracing_client),
    model="gemini-2.5-flash",
    instructions="You are a helpful assistant.",
    # Tracing is automatically enabled
)

# Run the agent - tracing happens automatically
response = agent.run(
    "What's the weather in Tokyo?", 
    trace_params={
        "name": "weather_query",
        "user_id": "user123",
        "metadata": {"source": "mobile_app"}
    }
)
```

<img width="100%" alt="Observability Dashboard" src="https://github.com/user-attachments/assets/94937238-405c-4011-83e2-147cec5cf3e7" />

<img width="100%" alt="Detailed Trace View" src="https://github.com/user-attachments/assets/c38429db-982c-4158-864f-f03e7118618e" />

**Automatic Scoring System** tracks:
- 🎯 **Model Tier Score** - Evaluates the capability tier of the model used
- 🔧 **Tool Usage Score** - Measures how effectively the agent uses available tools  
- 💰 **Token Efficiency Score** - Analyzes the balance between input and output tokens
- ⚡ **Cost Efficiency Score** - Tracks the cost-effectiveness of each generation

<img width="100%" alt="Trace Scores" src="https://github.com/user-attachments/assets/f0aab337-ead3-417b-97ef-0126c833d347" />


# WhatsApp Integration Guide (experimental, EvolutionAPI only)

This guide explains how to use Agentle's WhatsApp integration to build production-ready WhatsApp bots with Evolution API.

## Overview

Agentle's WhatsApp integration provides:

- **🚀 Easy Setup**: Simple configuration with Evolution API
- **📦 Session Management**: Production-ready session storage with Redis support
- **🔧 Flexible Architecture**: Pluggable providers for different WhatsApp APIs
- **📊 Built-in Observability**: Logging, error handling, and monitoring
- **🌐 Instant APIs**: Convert bots to REST APIs with automatic documentation
- **🛡️ Production Ready**: Error handling, retries, and graceful degradation

## Quick Start

### 1. Set Up Evolution API

1. Install and run Evolution API server
2. Create an instance in Evolution API
3. Get your API key and instance name

### 2. Basic Bot Example

```python
import asyncio
from agentle.agents.agent import Agent
from agentle.agents.whatsapp.whatsapp_bot import WhatsAppBot
from agentle.agents.whatsapp.providers.evolution.evolution_api_provider import EvolutionAPIProvider
from agentle.agents.whatsapp.providers.evolution.evolution_api_config import EvolutionAPIConfig
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Create agent
agent = Agent(
    name="My WhatsApp Bot",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You are a helpful assistant."
)

# Configure Evolution API
config = EvolutionAPIConfig(
    base_url="http://localhost:8080",
    instance_name="my-bot",
    api_key="your-api-key"
)

# Create provider and bot
provider = EvolutionAPIProvider(config)
bot = WhatsAppBot(agent, provider)

# Run a sync service (async version also available)
bot.start()
```

## Session Management

### In-Memory Sessions (Development)

```python
from agentle.sessions.in_memory_session_store import InMemorySessionStore
from agentle.sessions.session_manager import SessionManager

# Create session store
session_store = InMemorySessionStore[WhatsAppSession](
    cleanup_interval_seconds=300  # Cleanup every 5 minutes
)

# Create session manager
session_manager = SessionManager[WhatsAppSession](
    session_store=session_store,
    default_ttl_seconds=1800,  # 30 minutes
    enable_events=True
)

# Use with provider
provider = EvolutionAPIProvider(
    config=evolution_config,
    session_manager=session_manager
)
```

### Redis Sessions (Production)

```python
from agentle.sessions.redis_session_store import RedisSessionStore

# Create Redis session store
session_store = RedisSessionStore[WhatsAppSession](
    redis_url="redis://localhost:6379/0",
    key_prefix="whatsapp:sessions:",
    default_ttl_seconds=3600,  # 1 hour
    session_class=WhatsAppSession
)

# Create session manager
session_manager = SessionManager[WhatsAppSession](
    session_store=session_store,
    default_ttl_seconds=3600,
    enable_events=True
)

# Use with provider
provider = EvolutionAPIProvider(
    config=evolution_config,
    session_manager=session_manager
)
```

### Session Events

```python
# Add event handlers
async def on_session_created(session_id: str, session_data: Any) -> None:
    print(f"New session: {session_id}")

async def on_session_deleted(session_id: str, session_data: Any) -> None:
    print(f"Session deleted: {session_id}")

session_manager.add_event_handler("session_created", on_session_created)
session_manager.add_event_handler("session_deleted", on_session_deleted)
```

## Bot Configuration

```python
from agentle.agents.whatsapp.models.whatsapp_bot_config import WhatsAppBotConfig

bot_config = WhatsAppBotConfig(
    typing_indicator=True,
    typing_duration=3,
    auto_read_messages=True,
    session_timeout_minutes=30,
    max_message_length=4000,
    welcome_message="Hello! How can I help you?",
    error_message="Sorry, something went wrong. Please try again.",
)

bot = WhatsAppBot(agent, provider, config=bot_config)
```

## Adding Tools

```python
def get_weather(location: str) -> str:
    """Get weather information for a location."""
    # Your weather API implementation
    return f"Weather in {location}: Sunny, 25°C"

def book_appointment(date: str, time: str) -> str:
    """Book an appointment."""
    # Your booking system implementation
    return f"Appointment booked for {date} at {time}"

# Create agent with tools
agent = Agent(
    name="Assistant Bot",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="""You are a helpful assistant that can:
    - Provide weather information
    - Book appointments
    Always be friendly and helpful.""",
    tools=[get_weather, book_appointment]
)
```

## REST API Deployment

### Simple API Server

```python
# Convert bot to BlackSheep application
app = bot.to_blacksheep_app(
    webhook_path="/webhook/whatsapp",
    show_error_details=False  # Set to True for development
)

# Run with uvicorn
# DOCUMENTATION AUTOMATICALLY AVAILABLE AT localhost:8000/docs
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
```

### Production API with Custom Configuration

```python
from blacksheep import Application
from blacksheep.server.openapi.v3 import OpenAPIHandler
from openapidocs.v3 import Info

# Create custom OpenAPI documentation
docs = OpenAPIHandler(
    ui_path="/api/docs",
    info=Info(
        title="My WhatsApp Bot API",
        version="1.0.0",
        description="Production WhatsApp bot with AI capabilities"
    )
)

# Convert bot to application
app = bot.to_blacksheep_app(
    webhook_path="/webhooks/whatsapp",
    docs=docs,
    show_error_details=False
)

# Add custom middleware, CORS, etc.
# app.use_cors()
# app.use_authentication()

# Deploy
uvicorn.run(app, host="0.0.0.0", port=8000)
```

## Environment Configuration

### Environment Variables

```bash
# Evolution API Configuration
EVOLUTION_API_URL=http://localhost:8080
EVOLUTION_INSTANCE_NAME=my-bot
EVOLUTION_API_KEY=your-api-key

# Session Storage
REDIS_URL=redis://localhost:6379/0

# Bot Configuration
BOT_MODE=production  # simple, development, production
WEBHOOK_URL=https://your-domain.com/webhook/whatsapp
PORT=8000
DEBUG=false

# AI Provider
GOOGLE_API_KEY=your-google-api-key
```

### Configuration Class

```python
import os
from dataclasses import dataclass

@dataclass
class WhatsAppBotSettings:
    # Evolution API
    evolution_url: str = os.getenv("EVOLUTION_API_URL", "http://localhost:8080")
    evolution_instance: str = os.getenv("EVOLUTION_INSTANCE_NAME", "my-bot")
    evolution_api_key: str = os.getenv("EVOLUTION_API_KEY", "")
    
    # Session storage
    redis_url: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
    session_ttl: int = int(os.getenv("SESSION_TTL", "3600"))
    
    # Bot behavior
    webhook_url: str = os.getenv("WEBHOOK_URL", "")
    typing_duration: int = int(os.getenv("TYPING_DURATION", "3"))
    auto_read: bool = os.getenv("AUTO_READ_MESSAGES", "true").lower() == "true"
    
    # Server
    port: int = int(os.getenv("PORT", "8000"))
    debug: bool = os.getenv("DEBUG", "false").lower() == "true"

# Use configuration
settings = WhatsAppBotSettings()

evolution_config = EvolutionAPIConfig(
    base_url=settings.evolution_url,
    instance_name=settings.evolution_instance,
    api_key=settings.evolution_api_key
)
```

## Error Handling and Monitoring

### Custom Error Handling

```python
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Add custom webhook handlers
async def on_webhook_error(payload, error):
    logging.error(f"Webhook error: {error}")
    # Send to monitoring system
    
bot.add_webhook_handler(on_webhook_error)
```

### Health Checks

```python
# Add health check endpoint with blacksheep!
@app.router.post("/health")
async def health_check():
    stats = provider.get_stats()
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "stats": stats
    }
```

### Monitoring Stats

```python
# Get provider statistics
provider_stats = provider.get_stats()
print(f"Instance: {provider_stats['instance_name']}")
print(f"Active sessions: {provider_stats['session_stats']['store_stats']['total_sessions']}")

# Get session manager statistics
session_stats = session_manager.get_stats()
print(f"Session events enabled: {session_stats['events_enabled']}")
print(f"Default TTL: {session_stats['default_ttl_seconds']}s")
```

## Advanced Features

### Message Filtering

```python
class CustomWhatsAppBot(WhatsAppBot):
    async def handle_message(self, message: WhatsAppMessage) -> None:
        # Filter spam or unwanted messages
        if self.is_spam(message):
            return
        
        # Add custom preprocessing
        processed_message = self.preprocess_message(message)
        
        # Call parent handler
        await super().handle_message(processed_message)
    
    def is_spam(self, message: WhatsAppMessage) -> bool:
        # Your spam detection logic here...
        return False
    
    def preprocess_message(self, message: WhatsAppMessage) -> WhatsAppMessage:
        # Your preprocessing logic here...
        return message
```

## Deployment

### Docker Deployment

```dockerfile
FROM python:3.13-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "whatsapp_bot_example.py", "production"]
```

### Docker Compose with Redis

```yaml
version: '3.8'

services:
  whatsapp-bot:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379/0
      - EVOLUTION_API_URL=http://evolution:8080
      - BOT_MODE=production
    depends_on:
      - redis
      - evolution
    
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    
  evolution:
    image: atendai/evolution-api:latest
    ports:
      - "8080:8080"
    environment:
      - DATABASE_URL=postgresql://postgres:password@postgres:5432/evolution
    depends_on:
      - postgres
    
  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_DB=evolution
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  redis_data:
  postgres_data:
```

### Kubernetes Deployment

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: whatsapp-bot
spec:
  replicas: 3
  selector:
    matchLabels:
      app: whatsapp-bot
  template:
    metadata:
      labels:
        app: whatsapp-bot
    spec:
      containers:
      - name: whatsapp-bot
        image: your-registry/whatsapp-bot:latest
        ports:
        - containerPort: 8000
        env:
        - name: REDIS_URL
          value: "redis://redis-service:6379/0"
        - name: EVOLUTION_API_URL
          value: "http://evolution-service:8080"
        - name: BOT_MODE
          value: "production"
        resources:
          limits:
            memory: "512Mi"
            cpu: "500m"
          requests:
            memory: "256Mi"
            cpu: "250m"
---
apiVersion: v1
kind: Service
metadata:
  name: whatsapp-bot-service
spec:
  selector:
    app: whatsapp-bot
  ports:
  - port: 80
    targetPort: 8000
  type: LoadBalancer
```

## Best Practices

### 1. Security

- Use environment variables for sensitive configuration
- Implement webhook verification
- Rate limit API endpoints
- Use HTTPS in production
- Rotate API keys regularly

### 2. Performance

- Use Redis for session storage in production
- Implement proper caching strategies
- Monitor memory usage with session cleanup
- Use connection pooling for databases
- Implement graceful shutdowns

### 3. Reliability

- Implement retry logic for failed operations
- Use circuit breakers for external services
- Monitor webhook delivery
- Set up proper logging and alerting
- Implement health checks

### 4. Scalability

- Use horizontal scaling with load balancers
- Implement sticky sessions if needed
- Use distributed session storage (Redis)
- Monitor and tune session TTLs
- Implement proper resource limits

## Troubleshooting

### Common Issues

1. **Connection Errors**
   ```bash
   # Check Evolution API status
   curl http://localhost:8080/instance/fetchInstances
   
   # Check Redis connection
   redis-cli ping
   ```

2. **Session Issues**
   ```python
   # Check session statistics
   stats = session_manager.get_stats()
   print(f"Active sessions: {stats}")
   
   # List active sessions
   sessions = await session_manager.list_sessions()
   print(f"Session IDs: {sessions}")
   ```

3. **Webhook Problems**
   ```python
   # Verify webhook URL
   webhook_url = provider.get_webhook_url()
   print(f"Current webhook: {webhook_url}")
   
   # Re-register webhook
   await provider.set_webhook_url("https://your-domain.com/webhook")
   ```

### Debug Mode

```python
# Enable detailed logging
import logging
logging.basicConfig(level=logging.DEBUG)

# Enable agent debugging
agent = Agent(
    # ... other params ...
    debug=True
)

# Create development server with error details
app = bot.to_blacksheep_app(show_error_details=True)
```

### Monitoring

```python
# Add monitoring endpoints
@app.router.post("/metrics")
async def metrics():
    return {
        "provider_stats": provider.get_stats(),
        "session_stats": session_manager.get_stats(),
        "timestamp": datetime.now().isoformat()
    }

@app.router.post("/sessions")
async def list_sessions():
    sessions = await session_manager.list_sessions(include_metadata=True)
    return {"sessions": sessions}
```

## 🏗️ Real-World Examples

### 💬 Customer Support Agent

```python
from agentle.agents.agent import Agent
from agentle.agents.asgi.blacksheep.agent_to_blacksheep_application_adapter import AgentToBlackSheepApplicationAdapter
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

# Define support tools
def search_knowledge_base(query: str) -> str:
    """Search the support knowledge base."""
    # Implementation would search your KB
    return "Found solution: Reset password via email link"

def create_ticket(issue: str, priority: str = "medium") -> str:
    """Create a support ticket."""
    # Implementation would create ticket in your system
    return f"Ticket created with ID: SUPP-12345"

# Create support agent
support_agent = Agent(
    name="Support Hero",
    generation_provider=GoogleGenerationProvider(),
    model="gemini-2.5-flash",
    instructions="You are an empathetic customer support specialist.",
    tools=[search_knowledge_base, create_ticket],
    static_knowledge=["support_policies.pdf", "faq.md"]
)

# Deploy as API
api = AgentToBlackSheepApplicationAdapter().adapt(support_agent)
```

### 📊 Data Analysis Pipeline

```python
from agentle.agents.agent import Agent
from agentle.agents.agent_pipeline import AgentPipeline
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider

provider = GoogleGenerationProvider()

# Create specialized agents
data_cleaner = Agent(
    name="Data Cleaner",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You clean and preprocess data, handling missing values and outliers."
)

statistician = Agent(
    name="Statistician",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You perform statistical analysis and identify significant patterns."
)

visualizer = Agent(
    name="Visualizer",
    generation_provider=provider,
    model="gemini-2.5-flash",
    instructions="You create clear descriptions of data visualizations and insights."
)

# Build analysis pipeline
analysis_pipeline = AgentPipeline(
    agents=[data_cleaner, statistician, visualizer]
)

# Process data
result = analysis_pipeline.run("Analyze this sales data: Q1: $1.2M, Q2: $1.5M, Q3: $1.1M, Q4: $2.1M")
```

### 🌍 Multi-Provider Resilience

```python
from agentle.agents.agent import Agent
from agentle.generations.providers.failover.failover_generation_provider import FailoverGenerationProvider
from agentle.generations.providers.google.google_generation_provider import GoogleGenerationProvider
from agentle.generations.providers.openai.openai import OpenaiGenerationProvider
from agentle.generations.providers.cerebras.cerebras_generation_provider import CerebrasGenerationProvider

# Never go down - automatically failover between providers
resilient_provider = FailoverGenerationProvider(
    generation_providers=[
        GoogleGenerationProvider(),
        OpenaiGenerationProvider(api_key="your-openai-key"),
        CerebrasGenerationProvider(api_key="your-cerebras-key")
    ],
    shuffle=True
)

agent = Agent(
    name="Resilient Assistant",
    generation_provider=resilient_provider,
    # Use ModelKind instead of specific model names for better compatibility
    model="category_pro",  # Each provider maps this to their equivalent model
    instructions="You are a helpful assistant."
)
```

#### 🧠 Using ModelKind for Provider Abstraction

Agentle provides a powerful abstraction layer with `ModelKind` that decouples your code from specific provider model names:

```python
# Instead of hardcoding provider-specific model names:
agent = Agent(generation_provider=provider, model="gpt-4o")  # Only works with OpenAI

# Use ModelKind for provider-agnostic code:
agent = Agent(generation_provider=provider, model="category_pro")  # Works with any provider
```

**Benefits of ModelKind:**

- **Provider independence**: Write code that works with any AI provider
- **Future-proof**: When providers release new models, only mapping tables need updates
- **Capability-based selection**: Choose models based on capabilities, not names
- **Perfect for failover**: Each provider automatically maps to its equivalent model
- **Consistency**: Standardized categories