Webhooks e Canais Personalizados
Construa integrações de webhook e adaptadores de canal personalizados utilizando BaseChannel, FastAPI e despacho assíncrono.
Embora o chatflow-agent ofereça canais prontos para Telegram e WhatsApp, sistemas modernos frequentemente exigem a conexão de enxames multiagente com microsserviços internos, a API de eventos do Slack, bots do Discord ou webhooks de CRMs (Zendesk, Intercom).
Por meio da interface BaseChannel e do extra opcional webhook (fastapi>=0.110.0, uvicorn>=0.29.0), você pode implementar receptores de webhook prontos para produção em poucas linhas de código.
A Arquitetura do BaseChannel
Todos os canais de comunicação herdam de BaseChannel. Ele estabelece um ciclo de vida padronizado:
[ Webhook HTTP de Entrada ]
|
v
+-------------------------------------------------------------------+
| WebhookChannel Personalizado |
| 1. Autenticar Requisição (Chave de API ou Assinatura HMAC) |
| 2. Extrair session_id, message e metadata |
| 3. Encaminhar via dispatch_async(session_id, message, metadata) |
+---------------------------------+---------------------------------+
|
v
+---------------------------------+---------------------------------+
| chatflow-agent Runner |
| - Bloqueio de Sessão e Recuperação de Memória |
| - Orquestração Multiagente e Chamada de Ferramentas |
| - Persistência de Estado |
+---------------------------------+---------------------------------+
|
AgentResponse (text, data)
|
v
[ Resposta HTTP Imediata ou Callback Assíncrono de Webhook ]Métodos Principais do BaseChannel
from chatflow_agent.channels import BaseChannel
class MyChannel(BaseChannel):
"""Minimal channel definition."""
def run(self, **kwargs):
"""Start the listening service."""
passattach(runner: Runner) -> BaseChannel: Vincula uma instância de Runner multiagente ao canal.dispatch_async(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Encaminha a mensagem ao runner vinculado de forma assíncrona, garantindo o bloqueio de concorrência por sessão.dispatch(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Equivalente síncrono para ambientes de execução bloqueantes.split_message(text: str, max_len: int = 4096) -> list[str]: Utilitário para fragmentar respostas longas respeitando limites de plataformas de mensagens.
Implementando um Webhook de Produção com FastAPI
Abaixo está uma implementação completa e pronta para produção de um canal de webhook customizado 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: Verificação criptográfica de assinatura
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 diretamente para o 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)Executando o Serviço de Webhook
Conecte o canal ao seu agente e runner:
from chatflow_agent import Agent, Runner
from chatflow_agent.core.memory import SQLiteSessionStore
# 1. Definir o agente
assistant = Agent(
name="APIAssistant",
instructions="You are an automated API assistant.",
)
# 2. Configurar memória persistente
store = SQLiteSessionStore("webhook_sessions.db")
runner = Runner(starting_agent=assistant, session_store=store)
# 3. Instanciar o canal e vincular o runner
channel = GenericWebhookChannel(secret="super-secret-key")
channel.attach(runner)
if __name__ == "__main__":
channel.run(host="0.0.0.0", port=8080)Montagem em Aplicações FastAPI Existentes
Se o seu backend já utiliza uma aplicação FastAPI ou Starlette, não é necessário rodar um servidor isolado. Você pode acionar runner.run_async() diretamente em manipuladores de rotas existentes:
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"]
# Executar o agente diretamente
response = await runner.run_async(
session_id=session_id,
user_message=incoming_text,
metadata={"priority": payload.get("priority")},
)
return {"status": "processed", "solution": response.content}Melhores Práticas de Segurança
Lista de Verificação de Segurança para Produção
- Validação de Assinatura HMAC: Sempre valide assinaturas de entrada (ex.:
X-Hub-Signature-256,X-Slack-Signature) utilizandohmac.compare_digestpara prevenir ataques de temporização. - Tokens de Autenticação: Exija um cabeçalho de token de API ou Bearer token se o webhook tiver origem em microsserviços internos.
- Confirmação Imediata com Ferramentas Lentas: Caso seu agente invoque ferramentas demoradas, retorne imediatamente um
HTTP 200ao emissor do webhook e despache o turno do agente como uma tarefa em segundo plano (asyncio.create_task), enviando o resultado por meio de uma URL de callback externa.
WhatsApp Business Cloud API
Integração do WhatsApp para ambientes de produção usando Meta Cloud API, webhooks com FastAPI, verificação de assinaturas HMAC e gestão da janela de 24 horas.
Tratamento de Erros e Recuperação
Estratégias de tratamento de erros em produção no chatflow-agent, hierarquia de exceções do SDK, autocorreção de ferramentas e reversão transacional de turnos.