Webhooks y Canales Personalizados

Construye integraciones de webhook y adaptadores de canales personalizados utilizando BaseChannel, FastAPI y despacho asíncrono.

Aunque chatflow-agent ofrece canales integrados para Telegram y WhatsApp, los sistemas modernos a menudo requieren conectar enjambres multi-agente con microservicios internos, la API de eventos de Slack, bots de Discord o webhooks de CRMs (Zendesk, Intercom).

A través de la interfaz BaseChannel y el extra opcional webhook (fastapi>=0.110.0, uvicorn>=0.29.0), puedes implementar receptores de webhook listos para producción en unas pocas líneas de código.

La Arquitectura de `BaseChannel`

Todos los canales de comunicación heredan de BaseChannel. Establece un ciclo de vida estandarizado:

[ Webhook HTTP Entrante ]
           |
           v
+-------------------------------------------------------------------+
|                     WebhookChannel Personalizado                  |
|   1. Autenticar Petición (Clave de API o Firma HMAC)              |
|   2. Extraer session_id, message y metadata                       |
|   3. Reenviar vía dispatch_async(session_id, message, metadata)   |
+---------------------------------+---------------------------------+
                                  |
                                  v
+---------------------------------+---------------------------------+
|                       chatflow-agent Runner                       |
|   - Bloqueo de Sesión y Recuperación de Memoria                   |
|   - Orquestación Multi-Agente y Llamadas a Herramientas           |
|   - Persistencia de Estado                                        |
+---------------------------------+---------------------------------+
                                  |
                        AgentResponse (text, data)
                                  |
                                  v
[ Respuesta HTTP Inmediata o Callback de Webhook Asíncrono ]

Métodos Clave de `BaseChannel`

from chatflow_agent.channels import BaseChannel

class MyChannel(BaseChannel):
    """Minimal channel definition."""

    def run(self, **kwargs):
        """Start the listening service."""
        pass

attach(runner: Runner) -> BaseChannel: Vincula una instancia de Runner al canal.
dispatch_async(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Reenvía el mensaje al runner adjunto de forma asíncrona, asegurando el bloqueo de concurrencia por sesión.
dispatch(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Equivalente síncrono para entornos de ejecución bloqueantes.
split_message(text: str, max_len: int = 4096) -> list[str]: Utilidad para fragmentar respuestas salientes extensas respetando los límites de mensajería.

Implementación de un Webhook de Producción con FastAPI

A continuación se muestra una implementación completa y lista para producción de un canal de webhook personalizado utilizando FastAPI:

import hmac
import hashlib
from typing import Any
from fastapi import FastAPI, Header, HTTPException, Request
import uvicorn
from chatflow_agent import Agent, Runner
from chatflow_agent.channels.base import BaseChannel, ChannelError

class GenericWebhookChannel(BaseChannel):
    """Custom webhook receiver supporting HMAC verification and session routing."""

    def __init__(self, secret: str = "webhook-secret") -> None:
        super().__init__()
        self.secret = secret
        self.app = FastAPI(title="Agent Webhook Channel")
        self._setup_routes()

    def _setup_routes(self) -> None:
        @self.app.post("/api/v1/chat")
        async def handle_chat_webhook(
            request: Request,
            x_signature: str = Header(None, alias="X-Webhook-Signature"),
        ) -> dict[str, Any]:
            body_bytes = await request.body()

            # Opcional: Verificación criptográfica de firma
            if self.secret and x_signature:
                expected = hmac.new(
                    self.secret.encode("utf-8"),
                    body_bytes,
                    hashlib.sha256,
                ).hexdigest()
                if not hmac.compare_digest(expected, x_signature):
                    raise HTTPException(status_code=403, detail="Invalid signature")

            data = await request.json()
            session_id = data.get("session_id", "default-session")
            message = data.get("message", "")
            metadata = data.get("metadata", {})

            if not message:
                raise HTTPException(status_code=400, detail="Field 'message' is required")

            # Despachar directamente al runner vinculado
            try:
                response = await self.dispatch_async(session_id, message, metadata=metadata)
                return {
                    "session_id": session_id,
                    "reply": response.text,
                    "agent": response.agent_name,
                }
            except ChannelError as err:
                raise HTTPException(status_code=500, detail=str(err))

    def run(self, host: str = "0.0.0.0", port: int = 8080) -> None:
        """Start Uvicorn server."""
        self._ensure_runner_attached()
        uvicorn.run(self.app, host=host, port=port)

Ejecución del Servicio de Webhook

Conecta el canal a tu agente y runner:

from chatflow_agent import Agent, Runner
from chatflow_agent.core.memory import SQLiteSessionStore

# 1. Definir agente
assistant = Agent(
    name="APIAssistant",
    instructions="You are an automated API assistant.",
)

# 2. Configurar memoria persistente
store = SQLiteSessionStore("webhook_sessions.db")
runner = Runner(starting_agent=assistant, session_store=store)

# 3. Instanciar canal y vincular runner
channel = GenericWebhookChannel(secret="super-secret-key")
channel.attach(runner)

if __name__ == "__main__":
    channel.run(host="0.0.0.0", port=8080)

Montaje dentro de Aplicaciones FastAPI Existentes

Si tu backend ya cuenta con una aplicación FastAPI o Starlette, no requieres un servidor independiente. Puedes invocar runner.run_async() directamente dentro de tus rutas de API:

from fastapi import FastAPI, Depends
from chatflow_agent import Agent, Runner

app = FastAPI()

agent = Agent(name="RouterAgent", instructions="Triage support requests.")
runner = Runner(starting_agent=agent)

@app.post("/crm/webhook")
async def crm_webhook_endpoint(payload: dict):
    session_id = payload["customer_id"]
    incoming_text = payload["ticket_description"]

    # Ejecutar el agente directamente
    response = await runner.run_async(
        session_id=session_id,
        user_message=incoming_text,
        metadata={"priority": payload.get("priority")},
    )

    return {"status": "processed", "solution": response.content}

Buenas Prácticas de Seguridad

Lista de Verificación de Seguridad en Producción

Validación de Firmas HMAC: Verifica siempre firmas entrantes (ej. X-Hub-Signature-256, X-Slack-Signature) utilizando hmac.compare_digest para evitar ataques de temporización.
Tokens de Autenticación: Exige una cabecera de token de API o Bearer token si el webhook proviene de microservicios internos.
Confirmación Inmediata con Herramientas Lentas: Si tu agente invoca herramientas que demoran varios segundos, retorna de inmediato un HTTP 200 al emisor del webhook y despacha el turno del agente como tarea en segundo plano (asyncio.create_task), enviando el resultado mediante una URL de callback saliente.

Webhooks y Canales Personalizados

La Arquitectura de BaseChannelCopy Anchor Link

Métodos Clave de BaseChannelCopy Anchor Link

Implementación de un Webhook de Producción con FastAPICopy Anchor Link

Ejecución del Servicio de WebhookCopy Anchor Link

Montaje dentro de Aplicaciones FastAPI ExistentesCopy Anchor Link

Buenas Prácticas de SeguridadCopy Anchor Link