chatflow-agent

Build autonomous multi-agent swarms for WhatsApp, Telegram, and Webhooks in seconds without heavy dependency chains.
Construye equipos de agentes autonomos para WhatsApp, Telegram y Webhooks en segundos sin lidiar con dependencias monstruosas.

pip install chatflow-agent

English Documentation | Guía en Español | PyPI Package | Report Issue

Key Pillars (Why chatflow-agent?)

Ultra-Lightweight Core: Only 3 runtime dependencies (google-genai, pydantic, httpx). A tiny ~72 KB package footprint that installs in seconds on Docker, serverless, and cloud VPS environments.
Omnichannel Native: Production-ready connectors for WhatsApp (Official Meta Cloud API), Telegram, Webhooks (FastAPI), and interactive CLI.
Multi-Provider Freedom: Switch between Google Gemini, Google Gemma (100% free local via Ollama), xAI Grok, OpenAI (GPT-4o), and Anthropic Claude with a single parameter. Zero vendor lock-in.
Production-Hardened Shields: Native per-session concurrency locking (asyncio.Lock) prevents out-of-order race conditions from fast typers. Webhook anti-500 shield prevents Meta retry bombardment storms during LLM downtime.
Autonomous Peer Handoffs: Agents delegate conversations dynamically based on customer intent, without rigid state machines or complex graph builders.

Quickstart in 10 Seconds

import asyncio

from chatflow_agent import Agent, Runner

# 1. Define specialist agents

tech_agent = Agent(name="TechSupport", model="gemini-2.5-flash", instructions="Help with bugs.")

billing_agent = Agent(name="Billing", model="gemini-2.5-flash", instructions="Help with invoices.")

# 2. Receptionist agent with autonomous peer handoffs

receptionist = Agent(

    name="Receptionist",

    model="gemini-2.5-flash",

    instructions="Greet the customer and route them to TechSupport or Billing.",

    handoffs=[tech_agent, billing_agent],

)

async def main():

    runner = Runner(starting_agent=receptionist)

    result = await runner.run_async(

        session_id="user_whatsapp_1",

        user_message="Hi! I need help with an invoice error on my account.",

    )

    print(f"[{result.active_agent_name}]: {result.content}")

    # Output: [Billing]: I would be glad to help check your invoice. Could you share your account ID?

if __name__ == "__main__":

    asyncio.run(main())

English Documentation

Overview & Architecture

chatflow-agent is designed for developers who want the multi-agent power of OpenAI Swarm without being locked into a single provider, combined with turnkey connectors for messaging platforms like WhatsApp and Telegram.


[ WhatsApp / Telegram / Webhook / CLI ]

                   │

                   ▼

         ┌───────────────────┐

         │      Runner       │ ◄─── Session Memory (per-user history)

         └─────────┬─────────┘

                   │

         ┌─────────┴─────────┐

         ▼                   ▼

  ┌──────────────┐    ┌──────────────┐

  │ Triage Agent │───►│  Spec. Agent │ (Autonomous Peer Handoff)

  │ (Gemini/Grok)│    │ (Local Gemma)│

  └──────────────┘    └──────────────┘

         │                   │

         ▼                   ▼

    [@agent.tool]       [@agent.tool]

API Keys & Configuration

You can provide API keys using any of the following 3 methods:

1. Direct in Python Code (Easiest)

Pass your API key directly when instantiating the Runner or Agent:

# Pass to Runner (used by all agents with that provider)

runner = Runner(starting_agent=my_agent, api_key="AIzaSyYourGeminiKey")

# Or pass directly to a specific Agent:

grok_agent = Agent(

    name="GrokSpecialist",

    provider="grok",

    api_key="xai-your-key-here",

    instructions="..."

)

2. Using a `.env` File

Create a .env file in your project root:

# Google Gemini (Get free key at https://aistudio.google.com/)

GEMINI_API_KEY="AIzaSy..."

# OpenAI (https://platform.openai.com/api-keys)

OPENAI_API_KEY="sk-..."

# xAI Grok (https://console.x.ai/)

XAI_API_KEY="xai-..."

# Anthropic Claude (https://console.anthropic.com/)

ANTHROPIC_API_KEY="sk-ant-..."

Then in your script:

from dotenv import load_dotenv

load_dotenv()

3. Via Terminal Environment Variables

Windows (PowerShell):
```
$env:GEMINI_API_KEY="AIzaSy..."
```
Linux / macOS (Bash/Zsh):
```
export GEMINI_API_KEY="AIzaSy..."
```

Provider Credential Reference Table

Provider	Where to get Key	Environment Variable	In-Code Parameter	Free Tier Available?
Google Gemini	Google AI Studio	`GEMINI_API_KEY`	`api_key="..."`	Yes (Generous free tier)
Google Gemma (Local)	Ollama	None needed	`provider="ollama"`	100% Free & Offline
xAI Grok	xAI Console	`XAI_API_KEY`	`api_key="..."`	Pay-as-you-go
OpenAI	OpenAI Platform	`OPENAI_API_KEY`	`api_key="..."`	Pay-as-you-go
Anthropic Claude	Anthropic Console	`ANTHROPIC_API_KEY`	`api_key="..."`	Pay-as-you-go

Installation

Install only what you need:

# Core framework (Gemini, Gemma, Grok, OpenAI, Claude, CLI)

pip install chatflow-agent
`

# With WhatsApp Webhook connector (FastAPI + Uvicorn)

pip install "chatflow-agent[whatsapp]"

# With Telegram Bot connector (python-telegram-bot)

pip install "chatflow-agent[telegram]"

# Complete bundle (All channels & dev tools)

pip install "chatflow-agent[all]"

Step-by-Step Quickstarts

A. Cloud LLM Quickstart

Save as bot.py and run with python bot.py:

import asyncio

from chatflow_agent import Agent, Runner

# Define your agent

support_agent = Agent(

    name="SupportBot",

    model="gemini-2.5-flash",  # Or provider="openai", model="gpt-4o-mini"

    instructions="You are a helpful customer support agent for a retail store.",

)

# Register business tools using Python decorators and type hints

@support_agent.tool

def get_order_status(order_id: str) -> dict:

    """Look up shipping and tracking status for an order."""

    return {

        "order_id": order_id,

        "status": "Out for delivery",

        "carrier": "FedEx",

        "eta": "Today before 6:00 PM",

    }

async def main():

    # Pass api_key directly or set GEMINI_API_KEY in environment/.env

    runner = Runner(starting_agent=support_agent)

    response = await runner.run_async(

        session_id="user_session_101",

        user_message="Hi, can you check the status of my order #FDX-8821?",

    )

    print(f"[{response.active_agent_name}]: {response.content}")

if __name__ == "__main__":

    asyncio.run(main())

B. Local Offline Quickstart (Google Gemma)

Run 100% locally on your machine with zero API costs and total privacy using Ollama:

Start Ollama with Gemma: ollama run gemma2:9b
Run this script:

import asyncio

from chatflow_agent import Agent, Runner

# Connects to http://localhost:11434/v1 with no API key required

local_agent = Agent(

    name="LocalAnalyst",

    provider="ollama",

    model="gemma2:9b",

    instructions="You analyze confidential financial reports locally.",

)

@local_agent.tool

def calculate_vat(subtotal: float, rate_percentage: float = 21.0) -> dict:

    """Calculate VAT tax and total amount."""

    vat = subtotal * (rate_percentage / 100.0)

    return {"subtotal": subtotal, "vat": round(vat, 2), "total": round(subtotal + vat, 2)}

async def main():

    runner = Runner(starting_agent=local_agent)

    response = await runner.run_async(

        session_id="local_user",

        user_message="Calculate VAT for a $450 invoice.",

    )

    print(f"[{response.active_agent_name}]: {response.content}")

if __name__ == "__main__":

    asyncio.run(main())

C. Multi-Agent Swarm with Handoffs

Agents can autonomously delegate tasks to specialists:

import asyncio

from chatflow_agent import Agent, Runner

# 1. Specialist: Technical Support

tech_agent = Agent(

    name="TechSupport",

    model="gemini-2.5-flash",

    instructions="You diagnose hardware and software issues.",

)

@tech_agent.tool

def run_diagnostics(device_id: str) -> str:

    """Checks device telemetry."""

    return f"Device {device_id}: All sensors nominal. Firmware v2.1 up to date."

# 2. Specialist: Billing & Invoices

billing_agent = Agent(

    name="Billing",

    model="gemini-2.5-flash",

    instructions="You handle invoices, subscriptions, and refund requests.",

)

# 3. Receptionist (Frontline) with handoffs to specialists

concierge = Agent(

    name="Reception",

    model="gemini-2.5-flash",

    instructions="Greet customers and transfer to TechSupport or Billing as required.",

    handoffs=[tech_agent, billing_agent],  # Swarm handoff capability

)

async def main():

    runner = Runner(starting_agent=concierge)

    # The model detects it is a technical query and hands off to TechSupport automatically

    resp = await runner.run_async(

        session_id="client_77",

        user_message="My device DEV-42 is blinking red. Can you run diagnostics?",

    )

    print(f"Active Agent: {resp.active_agent_name}")  # Output: TechSupport

    print(f"Response: {resp.content}")

if __name__ == "__main__":

    asyncio.run(main())

Channel Connectors

WhatsApp Channel (Meta Cloud API)

Run a production webhook server compatible with Meta WhatsApp Business Cloud API:

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import WhatsAppChannel

agent = Agent(name="WhatsAppConcierge", instructions="Answer customer inquiries.")

runner = Runner(starting_agent=agent)

channel = WhatsAppChannel(

    verify_token="my_custom_webhook_secret",  # Verification token configured in Meta App

    access_token="EAA...",                    # Meta Permanent/System User Token

    app_secret="a1b2c3d4...",                 # Meta App Secret (App Dashboard -> App settings -> Basic)

    phone_number_id="109876543210987",        # WhatsApp Phone Number ID from Meta Dashboard

    fallback_message="We are experiencing a temporary delay. Please try again shortly.",

    unsupported_media_message="Currently I can only process text messages.",

)

channel.attach(runner)

if __name__ == "__main__":

    # Exposes GET /webhook (verification handshake) and POST /webhook (incoming messages)

    channel.run(host="0.0.0.0", port=8000)

Testing locally? Use a tunnel like ngrok (ngrok http 8000) or Cloudflare Tunnels to provide Meta with a public HTTPS URL (https://your-domain.ngrok-free.app/webhook).

Note

API Documentation in Development: For security, FastAPI interactive documentation (/docs, /redoc, and /openapi.json) is disabled by default in production. To enable Swagger UI and OpenAPI schemas during local development or debugging, pass enable_docs=True when instantiating WhatsAppChannel:

channel = WhatsAppChannel(
    ...,
    enable_docs=True,  # Enables /docs, /redoc, and /openapi.json for local development
)

Tip

Production Rate Limiting & Reverse Proxy: To protect /webhook and /health endpoints against flood attacks, denial of service (DoS), and CPU exhaustion from unauthenticated requests without risking dropping legitimate Meta webhook bursts at the application level, always place Uvicorn behind a reverse proxy in production (such as Nginx with limit_req_zone or Cloudflare Rate Limiting):

limit_req_zone $binary_remote_addr zone=whatsapp_limit:10m rate=10r/s;
location /webhook {
    limit_req zone=whatsapp_limit burst=20 nodelay;
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}

Telegram Channel

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import TelegramChannel

agent = Agent(name="TelegramBot", instructions="You assist Telegram users.")

runner = Runner(starting_agent=agent)

channel = TelegramChannel(

    token="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11",  # From @BotFather

    fallback_message="Sorry, a temporary issue occurred. Please retry in a few moments.",

)

channel.attach(runner)

if __name__ == "__main__":

    # Runs async polling; automatically handles /start, /reset, and typing indicators

    channel.run()

Interactive Terminal (CLI)

Test swarms in your terminal with colored chat output:

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import CLIChannel

agent = Agent(name="TerminalAssistant", instructions="Answer user questions concisely.")

runner = Runner(starting_agent=agent)

channel = CLIChannel(session_id="dev_test")

channel.attach(runner)

channel.run()

Production Resilience & Concurrency

ChatFlow includes built-in safeguards engineered specifically for real-world messaging traffic:

Per-Session Concurrency Lock (asyncio.Lock): When a user sends multiple messages in rapid succession (e.g., three WhatsApp voice transcriptions or quick texts in 2 seconds), an async lock guarantees FIFO execution. Turns are processed in strict sequential order, preventing race conditions, overlapping tool executions, or corrupted conversation histories.
Meta Anti-500 Error Shield: Meta's webhook infrastructure retries delivery aggressively if your endpoint returns HTTP 500. WhatsAppChannel catches any upstream LLM outages or rate limits, returns HTTP 200 to Meta to prevent retry storms, dispatches the configured fallback_message to the user, and logs clean diagnostic details.
Unsupported Media Handling: Audio voice notes, photos, and PDF files are safely intercepted with unsupported_media_message without throwing unhandled exceptions or disrupting ongoing chat sessions.
Turn Failure Rollback: If an LLM request times out, rate limits, or crashes mid-turn, ChatFlow automatically rolls back dialogue history to the last consistent turn in memory and SQLite (note: rollback reverts conversational history; external side effects of already executed tools are not rolled back), ensuring clean context without orphaned partial turns.

Session Memory & Persistence

ChatFlow provides native, zero-dependency persistence to ensure conversations and active specialist agents survive server restarts, worker redeployments, and crashes.

In-Memory Storage (`SessionStore`)

Ideal for testing, local scripting, and ephemeral sessions:

from chatflow_agent import Runner, SessionStore

runner = Runner(starting_agent=triage_agent, session_store=SessionStore(default_max_turns=20))

Persistent SQLite Storage (`SQLiteSessionStore`)

Built on Python's standard sqlite3 library with zero additional external packages. Safely manages Windows file locks and atomic transactions:

from chatflow_agent import Runner, SQLiteSessionStore

# History and active specialist agent state persist across server reboots

session_store = SQLiteSessionStore(db_path="sessions.db", default_max_turns=30)

runner = Runner(starting_agent=triage_agent, session_store=session_store)

Key Persistence Features:

Autonomous Handoff Survival: If a customer is handed off to Billing Specialist, the active agent pointer is recorded in SQLite. When the server reboots or workers recycle, subsequent customer messages continue directly with Billing Specialist instead of resetting to the triage agent.
Sliding Window FIFO Pruning: Maintains the latest conversation turns while safely discarding older messages in SQLite to avoid LLM context limits.
Omnichannel Compatibility: Works seamlessly across WhatsApp, Telegram, Webhook APIs, and CLI.

Production Starter Templates

Ready-to-run reference implementations are available in the examples/ directory:

examples/whatsapp_persistent_bot.py: Meta Cloud API WhatsApp bot with SQLite persistence, concurrency locks, and 3-agent handoff hierarchy.
examples/telegram_persistent_bot.py: Telegram bot with state persistence across polling restarts.
examples/webhook_api_service.py: Async REST API service with CORS, /api/chat and /api/sessions/{session_id} endpoints for React, Next.js, and mobile frontends.
examples/local_gemma_cli.py: 100% offline, free CLI assistant powered by Google Gemma 2 via Ollama with persistent terminal sessions.
examples/multi_provider_swarm.py: Multi-provider swarm coordinating Gemini, Gemma, Grok, and Claude in a single conversational flow.

Guía Completa en Español

Pilares Clave (Por que chatflow-agent?)

Nucleo Ultra-Ligero: Solo 3 dependencias de ejecucion (google-genai, pydantic, httpx). Un paquete diminuto de ~72 KB que se instala en segundos en contenedores Docker, entornos serverless y servidores VPS.
Nativo Omnicanal: Conectores listos para produccion para WhatsApp (Meta Cloud API Oficial), Telegram, Webhooks (FastAPI) y CLI interactiva.
Libertad Multi-Proveedor: Alterna entre Google Gemini, Google Gemma (100% gratis y local via Ollama), xAI Grok, OpenAI (GPT-4o) y Anthropic Claude con un unico parametro. Cero ataduras a proveedores.
Escudos Reforzados para Produccion: Bloqueo de concurrencia nativo por sesion (asyncio.Lock) que previene condiciones de carrera por usuarios que escriben rapido. Escudo anti-500 en webhooks para evitar tormentas de reintentos de Meta durante caidas del LLM.
Handoffs Autonomos entre Agentes: Los agentes delegan conversaciones dinamicamente segun la intencion del usuario, sin maquinas de estado rigidas ni grafos complejos.

Inicio Rapido en 10 Segundos

import asyncio
from chatflow_agent import Agent, Runner

# 1. Definir agentes especialistas
tech_agent = Agent(name="TechSupport", model="gemini-2.5-flash", instructions="Ayuda con errores tecnicos.")
billing_agent = Agent(name="Billing", model="gemini-2.5-flash", instructions="Ayuda con facturacion y pagos.")

# 2. Agente recepcionista con handoffs autonomos
receptionist = Agent(
    name="Receptionist",
    model="gemini-2.5-flash",
    instructions="Saluda cordialmente al cliente y derivalo a TechSupport o Billing segun su consulta.",
    handoffs=[tech_agent, billing_agent],
)

async def main():
    runner = Runner(starting_agent=receptionist)
    result = await runner.run_async(
        session_id="usuario_whatsapp_1",
        user_message="Hola! Necesito ayuda con un error en la factura de mi cuenta.",
    )
    print(f"[{result.active_agent_name}]: {result.content}")
    # Salida: [Billing]: Con gusto te ayudo a revisar tu factura. ¿Me podrias indicar tu numero de cuenta?

if __name__ == "__main__":
    asyncio.run(main())

Tabla de Contenidos

Vision General y Arquitectura
Configuracion de API Keys y Credenciales
Instalacion
Guias Paso a Paso (Quickstarts)
Conectores de Canales Oficiales
Resiliencia y Concurrencia en Produccion
Memoria de Sesiones y Persistencia en SQLite
Plantillas de Produccion Listas para Usar
Documentación en Inglés

Vision General y Arquitectura

chatflow-agent fue disenado para desarrolladores que buscan la potencia multi-agente de OpenAI Swarm sin quedar atados a un unico proveedor, combinado con conectores oficiales listos para produccion en plataformas de mensajeria como WhatsApp y Telegram.


[ WhatsApp / Telegram / Webhook / CLI ]

                   │

                   ▼

         ┌───────────────────┐

         │      Runner       │ ◄─── Memoria de Sesion (Historial por usuario)

         └─────────┬─────────┘

                   │

         ┌─────────┴─────────┐

         ▼                   ▼

  ┌──────────────┐    ┌──────────────┐

  │ Agente Triage│───►│Especialista  │ (Handoff Autonomo entre Pares)

  │ (Gemini/Grok)│    │(Gemma Local) │

  └──────────────┘    └──────────────┘

         │                   │

         ▼                   ▼

    [@agent.tool]       [@agent.tool]

Configuracion de API Keys y Credenciales

Puedes configurar tus credenciales utilizando cualquiera de estos 3 metodos:

Opcion 1: Directo en tu Codigo Python (La mas rapida)

Pasa tu clave al instanciar el Runner o el Agent:

# Pasándola al Runner (la usan todos los agentes de ese proveedor):

runner = Runner(starting_agent=mi_agente, api_key="AIzaSyTuClaveDeGemini")

# O a un agente especifico:

agente_grok = Agent(

    name="GrokSpecialist",

    provider="grok",

    api_key="xai-tu-clave-aqui",

    instructions="Eres un analista de datos.",

)

Opcion 2: Usando un Archivo `.env` (Recomendado en Produccion)

Crea un archivo .env en la raiz de tu proyecto:

# Google Gemini (Obtén tu clave gratuita en https://aistudio.google.com/)

GEMINI_API_KEY="AIzaSy..."

# OpenAI (https://platform.openai.com/api-keys)

OPENAI_API_KEY="sk-..."

# xAI Grok (https://console.x.ai/)

XAI_API_KEY="xai-..."

# Anthropic Claude (https://console.anthropic.com/)

ANTHROPIC_API_KEY="sk-ant-..."

Opcion 3: Variables de Entorno en la Terminal

Windows (PowerShell):
```
$env:GEMINI_API_KEY="AIzaSy..."
```
Linux / macOS (Bash/Zsh):
```
export GEMINI_API_KEY="AIzaSy..."
```

Tabla Comparativa de Proveedores y Credenciales

Proveedor	Donde obtener la Clave	Variable de Entorno	Parametro en Codigo	Nivel Gratuito Disponible
Google Gemini	Google AI Studio	`GEMINI_API_KEY`	`api_key="..."`	Si (Nivel gratuito generoso)
Google Gemma (Local)	Ollama	No requiere clave	`provider="ollama"`	100% Gratis y Offline
xAI Grok	Consola de xAI	`XAI_API_KEY`	`api_key="..."`	Pago por uso
OpenAI	Plataforma OpenAI	`OPENAI_API_KEY`	`api_key="..."`	Pago por uso
Anthropic Claude	Consola Anthropic	`ANTHROPIC_API_KEY`	`api_key="..."`	Pago por uso

Instalacion

Instala unicamente los componentes que tu proyecto requiera:

# Nucleo base (Gemini, Gemma, Grok, OpenAI, Claude, CLI)

pip install chatflow-agent
`

# Con conector de Webhooks para WhatsApp (FastAPI + Uvicorn)

pip install "chatflow-agent[whatsapp]"

# Con conector de Bot para Telegram (python-telegram-bot)

pip install "chatflow-agent[telegram]"

# Paquete completo (Todos los canales y herramientas de desarrollo)

pip install "chatflow-agent[all]"

Guias Paso a Paso (Quickstarts)

A. Agente Cloud con Herramientas

Guarda este codigo como asistente.py y ejecutalo con python asistente.py:

import asyncio

from chatflow_agent import Agent, Runner

# 1. Definir el agente

agente_soporte = Agent(

    name="SoporteTienda",

    model="gemini-2.5-flash",  # O provider="openai", model="gpt-4o-mini"

    instructions="Eres un asistente cordial de atencion al cliente para una tienda online.",

)

# 2. Registrar herramientas de negocio usando decoradores y tipado estándar

@agente_soporte.tool

def consultar_pedido(pedido_id: str) -> dict:

    """Consulta el estado de despacho y envio de un pedido por su identificador."""

    return {

        "pedido_id": pedido_id,

        "estado": "En distribucion",

        "transporte": "Envios Express",

        "llegada_estimada": "Hoy antes de las 18:00 hs",

    }

async def main():

    runner = Runner(starting_agent=agente_soporte)

    respuesta = await runner.run_async(

        session_id="cliente_101",

        user_message="Hola, ¿podrias revisar en que estado se encuentra mi pedido #PED-9921?",

    )

    print(f"[{respuesta.active_agent_name}]: {respuesta.content}")

if __name__ == "__main__":

    asyncio.run(main())

B. Agente 100% Offline y Gratuito (Google Gemma)

Ejecuta agentes de forma local en tu propia computadora, sin costos de API y con total privacidad de datos usando Ollama:

Inicia Ollama con Gemma: ollama run gemma2:2b (o gemma2:9b)
Ejecuta este script:

import asyncio

from chatflow_agent import Agent, Runner

# Se conecta a http://localhost:11434/v1 sin requerir claves de API

agente_local = Agent(

    name="AnalistaLocal",

    provider="ollama",

    model="gemma2:2b",

    instructions="Eres un analista financiero que procesa informacion confidencial de manera local.",

)

@agente_local.tool

def calcular_iva(subtotal: float, tasa_porcentaje: float = 21.0) -> dict:

    """Calcula el impuesto de IVA y el monto total de una operacion comercial."""

    iva = subtotal * (tasa_porcentaje / 100.0)

    return {"subtotal": subtotal, "iva": round(iva, 2), "total": round(subtotal + iva, 2)}

async def main():

    runner = Runner(starting_agent=agente_local)

    respuesta = await runner.run_async(

        session_id="usuario_local",

        user_message="Calcula el IVA de una factura de 550 dolares al 21%.",

    )

    print(f"[{respuesta.active_agent_name}]: {respuesta.content}")

if __name__ == "__main__":

    asyncio.run(main())

C. Swarm Multi-Agente con Handoffs Autonomos

Los agentes pueden delegar la conversacion a especialistas de forma autonoma segun la intencion del usuario:

import asyncio

from chatflow_agent import Agent, Runner

# 1. Especialista: Soporte Tecnico

agente_tecnico = Agent(

    name="SoporteTecnico",

    model="gemini-2.5-flash",

    instructions="Resuelves incidentes tecnicos de conectividad y servidores.",

)

@agente_tecnico.tool

def diagnosticar_nodo(nodo_id: str) -> dict:

    """Diagnostica el estado de un nodo de red."""

    return {"nodo_id": nodo_id, "estado": "Operativo", "latencia_ms": 22}

# 2. Especialista: Facturacion y Cobranzas

agente_facturacion = Agent(

    name="Facturacion",

    model="gemini-2.5-flash",

    instructions="Resuelves dudas sobre cobros, medios de pago y recibos.",

)

# 3. Recepcionista de Primera Linea (Frontline) con handoffs a especialistas

recepcionista = Agent(

    name="Recepcion",

    model="gemini-2.5-flash",

    instructions="Saludas al usuario y derivas a SoporteTecnico o Facturacion segun corresponda.",

    handoffs=[agente_tecnico, agente_facturacion],

)

async def main():

    runner = Runner(starting_agent=recepcionista)

    # El modelo detecta que es un problema tecnico y realiza la transferencia de forma automatica

    respuesta = await runner.run_async(

        session_id="cliente_77",

        user_message="Tengo una caida en el nodo SRV-04. ¿Pueden revisarlo?",

    )

    print(f"Agente Activo: {respuesta.active_agent_name}")  # Salida: SoporteTecnico

    print(f"Respuesta: {respuesta.content}")

if __name__ == "__main__":

    asyncio.run(main())

Conectores de Canales Oficiales

Canal de WhatsApp (Meta Cloud API)

Servidor de webhook listo para produccion, compatible con la API oficial de WhatsApp Business en Meta Cloud:

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import WhatsAppChannel

agente_ventas = Agent(name="ConciergeWhatsApp", instructions="Atiende consultas comerciales.")

runner = Runner(starting_agent=agente_ventas)

canal_whatsapp = WhatsAppChannel(

    verify_token="tu_token_verificacion_meta", # Configurado en tu app de Meta Developers

    access_token="EAA...",                     # Token permanente de System User en Meta

    app_secret="a1b2c3d4...",                  # Meta App Secret (Panel de Meta -> Configuración -> Básica)

    phone_number_id="109876543210987",         # WhatsApp Phone Number ID del panel de Meta

    fallback_message="Disculpa, estamos experimentando una demora temporal. Por favor intenta en unos momentos.",

    unsupported_media_message="Por el momento solo puedo procesar mensajes de texto.",

)

canal_whatsapp.attach(runner)

if __name__ == "__main__":

    # Expone GET /webhook (verificacion de Meta) y POST /webhook (recepcion de mensajes)

    canal_whatsapp.run(host="0.0.0.0", port=8000)

¿Probando localmente? Usa un tunel como ngrok (ngrok http 8000) o Cloudflare Tunnels para brindarle a Meta la URL publica HTTPS (https://tu-dominio.ngrok-free.app/webhook).

Note

Documentación interactiva en desarrollo: Por motivos de seguridad, la documentación interactiva de FastAPI (/docs, /redoc y /openapi.json) está deshabilitada por defecto en producción. Para habilitar Swagger UI y los esquemas OpenAPI durante el desarrollo local o debugging, pasa enable_docs=True al instanciar WhatsAppChannel:

canal_whatsapp = WhatsAppChannel(
    ...,
    enable_docs=True,  # Habilita /docs, /redoc y /openapi.json para desarrollo local
)

Tip

Rate Limiting y Reverse Proxy en Producción: Para proteger los endpoints /webhook y /health contra ataques de inundación, denegación de servicio (DoS) y agotamiento de CPU por peticiones no autenticadas sin riesgo de descartar ráfagas legítimas de webhooks de Meta a nivel aplicación, ubica siempre Uvicorn detrás de un reverse proxy en producción (como Nginx con limit_req_zone o Cloudflare Rate Limiting):

limit_req_zone $binary_remote_addr zone=whatsapp_limit:10m rate=10r/s;
location /webhook {
    limit_req zone=whatsapp_limit burst=20 nodelay;
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}

Canal de Telegram

Conecta tu enjambre de agentes a un Bot de Telegram con polling asincrono y soporte nativo de comandos:

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import TelegramChannel

agente_bot = Agent(name="TelegramBot", instructions="Asistes a los usuarios en Telegram.")

runner = Runner(starting_agent=agente_bot)

canal_telegram = TelegramChannel(

    token="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11",  # Token provisto por @BotFather

    fallback_message="Disculpa, ocurrio un inconveniente temporal. Por favor reintenta en breve.",

    start_message="Bienvenido al bot. ¿En que podemos ayudarte hoy?",

)

canal_telegram.attach(runner)

if __name__ == "__main__":

    # Ejecuta el bucle asincrono; gestiona /start, /reset e indicadores de escritura de forma automatica

    canal_telegram.run()

Canal de Terminal Interactiva (CLI)

Prueba y depura enjambres de agentes directamente en tu consola con formato legible:

from chatflow_agent import Agent, Runner

from chatflow_agent.channels import CLIChannel

agente_terminal = Agent(name="AsistenteCLI", instructions="Responde de manera concisa y tecnica.")

runner = Runner(starting_agent=agente_terminal)

canal_cli = CLIChannel(session_id="prueba_desarrollo")

canal_cli.attach(runner)

canal_cli.run()

Resiliencia y Concurrencia en Produccion

chatflow-agent incorpora protecciones nativas disenadas especificamente para el trafico real de canales de mensajeria:

Candado de Concurrencia por Sesion (asyncio.Lock): Si un usuario envia varios mensajes consecutivos en pocos segundos (mensajes rapidos o transcripciones de notas de voz), un cerrojo asincrono garantiza el procesamiento en estricto orden FIFO. Esto evita condiciones de carrera, ejecuciones cruzadas de herramientas y corrupcion de historiales.
Escudo Anti-500 para Webhooks de Meta: Si tu endpoint devuelve un error 500, los servidores de Meta reintentan agresivamente la entrega, saturando el servidor. WhatsAppChannel captura caidas o limites de cuota de los LLMs, responde HTTP 200 a Meta para detener la tormenta de reintentos, envia al cliente un mensaje amigable (fallback_message) y registra el diagnostico de forma limpia.
Manejo Seguro de Archivos Multimedia: Audios, fotos y documentos son interceptados con un aviso claro (unsupported_media_message) sin generar excepciones no controladas ni interrumpir la sesion en curso.
Rollback ante Fallas en el Turno: Si una solicitud al LLM excede el tiempo de espera o falla a mitad de turno, ChatFlow revierte automaticamente el historial de conversacion al ultimo turno consistente en memoria y SQLite (nota: revierte el historial conversacional; no deshace efectos secundarios externos de tools ya ejecutadas), garantizando un contexto limpio sin turnos huerfanos.

Memoria de Sesiones y Persistencia en SQLite

chatflow-agent ofrece persistencia nativa con cero dependencias externas adicionales, garantizando que las conversaciones y los agentes especialistas activos sobrevivan a reinicios del servidor o reciclado de procesos.

Almacenamiento Volatil en Memoria RAM (`SessionStore`)

Ideal para pruebas, scripts locales o sesiones efimeras:

from chatflow_agent import Runner, SessionStore

runner = Runner(starting_agent=agente_recepcion, session_store=SessionStore(default_max_turns=20))

Almacenamiento Persistente en Disco SQLite (`SQLiteSessionStore`)

Basado en la libreria estandar sqlite3 de Python. Gestiona de manera segura transacciones atomicas y bloqueos de archivo en Windows:

from chatflow_agent import Runner, SQLiteSessionStore

# El historial y la derivacion al especialista activo se conservan entre reinicios del servidor

session_store = SQLiteSessionStore(db_path="conversaciones.db", default_max_turns=30)

runner = Runner(starting_agent=agente_recepcion, session_store=session_store)

Ventajas clave:

Persistencia del Agente Activo: Si un cliente fue derivado a Facturacion, el puntero se guarda en SQLite. Al reiniciar el servidor, los nuevos mensajes continuan directamente con Facturacion sin volver a comenzar en la recepcion.
Poda por Ventana Deslizante (FIFO): Mantiene acotado el tamano del historial en la base de datos para no desbordar la ventana de contexto de los modelos.
Compatibilidad Omnicanal: Funciona exactamente igual en WhatsApp, Telegram, REST APIs y CLI.

Plantillas de Produccion Listas para Usar

En el directorio examples/ encontraras proyectos completos listos para clonar y ejecutar en tu entorno:

examples/whatsapp_persistent_bot.py: Bot de WhatsApp Business (Meta Cloud API) con persistencia SQLite, candados de concurrencia y jerarquia de 3 agentes con handoffs.
examples/telegram_persistent_bot.py: Bot de Telegram con memoria de sesion y retencion de especialista entre reinicios de polling.
examples/webhook_api_service.py: Servicio REST API asincrono con CORS y endpoints /api/chat y /api/sessions/{session_id} listo para frontends en React, Next.js y apps moviles.
examples/local_gemma_cli.py: Asistente de terminal 100% offline y gratuito con Google Gemma 2 via Ollama y sesiones persistentes en SQLite.
examples/multi_provider_swarm.py: Enjambre multi-proveedor que coordina Gemini, Gemma local, Grok y Claude en un unico flujo conversacional.

Author & Community / Autor y Comunidad

Created and maintained by / Creado y mantenido por Gustavo Baranda.

If you find chatflow-agent useful for your projects, consider giving it a star on GitHub!
Si encuentras util chatflow-agent para tus proyectos, considera darle una estrella en GitHub!

Contributions, issues, and feature requests are always welcome.
Contribuciones, reportes de errores y solicitudes de caracteristicas son siempre bienvenidos.

License / Licencia

Distributed under the MIT License / Distribuido bajo la Licencia MIT.

Name		Name	Last commit message	Last commit date
.github/workflows		.github/workflows
examples		examples
src/chatflow_agent		src/chatflow_agent
tests		tests
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml

Latest commit

History

Folders and files

Repository files navigation