Webhook & Custom Channels
Build custom webhook integrations and channel adapters using BaseChannel, FastAPI, and asynchronous dispatching.
While chatflow-agent provides pre-built channels for Telegram and WhatsApp, modern systems frequently require connecting multi-agent swarms to internal microservices, Slack Events APIs, Discord bots, or CRM webhooks (Zendesk, Intercom).
Through the BaseChannel interface and the optional webhook extra (fastapi>=0.110.0, uvicorn>=0.29.0), you can implement production-grade webhook listeners in just a few lines of code.
The BaseChannel Architecture
All communication channels inherit from BaseChannel. It establishes a standardized lifecycle:
[ Incoming HTTP Webhook ]
|
v
+-------------------------------------------------------------------+
| Custom WebhookChannel |
| 1. Authenticate Request (API Key or HMAC Signature) |
| 2. Extract session_id, message, and metadata |
| 3. Forward via dispatch_async(session_id, message, metadata) |
+---------------------------------+---------------------------------+
|
v
+---------------------------------+---------------------------------+
| chatflow-agent Runner |
| - Session Lock & Memory Retrieval |
| - Multi-Agent Orchestration & Tool Calling |
| - State Persistence |
+---------------------------------+---------------------------------+
|
AgentResponse (text, data)
|
v
[ Immediate HTTP Response or Async Webhook Callback ]Key Methods of 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: Binds a multi-agent Runner instance to the channel.dispatch_async(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Forwards the message to the attached runner asynchronously, ensuring session concurrency locking.dispatch(session_id: str, message: str, metadata: dict | None = None) -> AgentResponse: Synchronous counterpart for synchronous runtimes.split_message(text: str, max_len: int = 4096) -> list[str]: Utility for chunking long outbound responses across messaging limits.
Implementing a Production FastAPI Webhook
Below is a complete, production-ready implementation of a custom webhook channel using 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()
# Optional: Cryptographic signature verification
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")
# Dispatch directly into the attached runner
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)Running the Webhook Service
Connect the channel to your agent and runner:
from chatflow_agent import Agent, Runner
from chatflow_agent.core.memory import SQLiteSessionStore
# 1. Define agent
assistant = Agent(
name="APIAssistant",
instructions="You are an automated API assistant.",
)
# 2. Configure persistent memory
store = SQLiteSessionStore("webhook_sessions.db")
runner = Runner(starting_agent=assistant, session_store=store)
# 3. Instantiate channel and attach runner
channel = GenericWebhookChannel(secret="super-secret-key")
channel.attach(runner)
if __name__ == "__main__":
channel.run(host="0.0.0.0", port=8080)Mounting Inside Existing FastAPI Applications
If your backend already runs a FastAPI or Starlette application, you don't need a separate server. You can mount runner.run_async() directly inside existing API route handlers:
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"]
# Execute agent directly
response = await runner.run_async(
session_id=session_id,
user_message=incoming_text,
metadata={"priority": payload.get("priority")},
)
return {"status": "processed", "solution": response.content}Security Best Practices
Production Security Checklist
- HMAC Signature Validation: Always verify incoming signatures (e.g.
X-Hub-Signature-256,X-Slack-Signature) usinghmac.compare_digestto prevent timing attacks. - Authentication Tokens: Require an API token or Bearer token header if the webhook originates from internal microservices.
- Immediate Acknowledgement on Slow Tools: If your agent invokes long-running tools, return an immediate
HTTP 200to the webhook provider and dispatch the agent turn as a background task (asyncio.create_task), sending the result via an outbound callback URL.
WhatsApp Business Cloud API
Production-grade WhatsApp integration using Meta Cloud API, FastAPI webhooks, HMAC signature verification, and 24-hour window management.
Error Handling & Recovery
Production error handling strategies in chatflow-agent, SDK exception hierarchy, automatic tool self-correction, and transactional turn rollbacks.