A team wires per-tool webhook dispatch from their LangChain agent via FastAPI
BackgroundTasks — analytics is always N seconds late because BackgroundTasks
fire after the HTTP response closes, not during the stream (P60). Worse:
the BaseCallbackHandler they attached via .with_config(callbacks=[h])
fires on the outer agent but is dark on the subagent's tool calls — custom
callbacks are not inherited by LangGraph subgraphs (P28), they must be
passed via config["callbacks"] at invoke time.
Pain-catalog anchors handled here:
P28 — Callbacks via with_config don't propagate to subgraphs
P46 — SSE streams dropped by buffering proxies (see langchain-langgraph-streaming)
P47 — astream_events(v2) emits thousands of events; never forward raw
P60 — BackgroundTasks fire post-response; wrong for per-event dispatch
This skill walks through an async AsyncCallbackHandler with fire-and-forget
dispatch, per-target sinks for HTTP / Kafka / Redis Streams / SNS, HMAC-signed
delivery with 1s/5s/30s retry and DLQ, idempotency keys = `run_id + event_type
step_index, andconfig["callbacks"]wiring that makes subagent calls visible. Typical webhook latency budget: <500ms per event. Pin:langchain-core 1.0.x, langgraph 1.0.x. Scope: server-to-server dispatch only — UI streaming is in langchain-langgraph-streaming`.
One of: aiokafka, redis[hiredis] >= 5, aioboto3 (per target)
An event sink — a webhook endpoint, Kafka topic, Redis Stream, or SNS topic
A shared secret (for HMAC) stored in your secret manager, not env
Instructions
Step 1 — Write an async handler that fire-and-forget dispatches
Sync dispatch from a callback blocks the chain — a slow HTTP POST during
on_tool_end serializes all downstream tokens behind it (P48). Use
so the dispatch runs alongside the chain:
asyncio.create_task(...)
import asyncio
import uuid
from typing importAnyfrom langchain_core.callbacks import AsyncCallbackHandler
classEventDispatchHandler(AsyncCallbackHandler):
"""Fire-and-forget dispatch to external sinks.
IMPORTANT: subclass AsyncCallbackHandler (not BaseCallbackHandler) so
on_* methods are awaited. Mixing sync and async handlers is a silent
footgun — sync on_* blocks the event loop (P48).
"""def__init__(self, sink, *, run_id: str | None = None):
self.sink = sink # dispatch target — Step 3self.run_id = run_id orstr(uuid.uuid4())
self._tasks: set[asyncio.Task] = set()
def_dispatch(self, event_type: str, payload: dict, step_index: int) -> None:
# Fire-and-forget. Keep a strong reference so the task isn't GC'd# mid-flight (asyncio quirk — orphan tasks get garbage-collected).
task = asyncio.create_task(
self.sink.send(
idempotency_key=f"{self.run_id}:{event_type}:{step_index}",
event_type=event_type,
payload=payload,
)
)
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)
asyncdefon_tool_end(self, output: Any, *, run_id, parent_run_id=None, **kwargs):
# 4000 char cap — keep payload under typical webhook body limits# while preserving enough context for downstream analytics.
MAX_OUTPUT_CHARS = 4000self._dispatch(
"tool_end",
{"output": str(output)[:MAX_OUTPUT_CHARS], "run_id": str(run_id)},
step_index=kwargs.get("tags", []).__len__() or0,
)
asyncdefon_chain_end(self, outputs: dict, *, run_id, **kwargs):
# Only named chains — skip the unnamed LCEL inner nodes (P47)
name = kwargs.get("name")
ifnot name or name.startswith("RunnableLambda"):
returnself._dispatch("chain_end", {"name": name, "run_id": str(run_id)}, step_index=0)
asyncdefdrain(self, timeout: float = 5.0) -> None:
"""Call before process exit so in-flight dispatches complete."""ifself._tasks:
await asyncio.wait(self._tasks, timeout=timeout)
See Async Callback Handler for the full
handler — on_llm_end, filtering, sync-vs-async decision.
Step 2 — Pass callbacks via config so subgraphs inherit them
P28: Runnable.with_config(callbacks=[h]) binds at definition and is not
inherited by LangGraph subgraphs. Pass callbacks via config at invocation:
# WRONG — subagent tool calls never fire the handler
agent_with_handler = agent.with_config({"callbacks": [handler]})
await agent_with_handler.ainvoke({"messages": [...]})
# RIGHT — callbacks in config propagate into subgraphsawait agent.ainvoke(
{"messages": [...]},
config={"callbacks": [handler], "configurable": {"thread_id": "t1"}},
)
Validate propagation with a probe that counts events by kwargs["name"] and
asserts the subagent's name appears. See Subgraph Propagation.
Step 3 — Pick a dispatch target by delivery semantics
See Dispatch Targets for Kafka / Redis Streams
/ SNS sinks and per-target DLQ patterns.
Step 4 — Filter events so you don't saturate the downstream
astream_events(version="v2") emits thousands of events per invocation (P47).
Never forward raw — dispatch only what the downstream consumes:
Callback method
Typical decision
Why
on_llm_start
Skip
Prompt content often contains PII; low value without masking
on_llm_new_token
Skip for dispatch (UI only)
1 event per token; N/A to analytics
on_llm_end
Dispatch for named chains only
Token usage, final response — high value, low volume
on_chain_start
Skip (P47 noise)
LCEL emits one per inner runnable
on_chain_end
Dispatch for named subgraphs only
Stage completion — what analytics cares about
on_tool_start
Optional (dispatch for audit log)
Matters for compliance / tool-use audit
on_tool_end
Dispatch always
The key analytics signal in agent flows
on_agent_action
Dispatch
Cleaner signal than on_tool_start in agent graphs
on_agent_finish
Dispatch
Terminal event for the run
Rule of thumb: dispatch on_tool_end + named on_chain_end + on_llm_end.
Everything else is noise.
Step 5 — Build idempotency keys and a retry budget
At-least-once transports mean duplicates. Build the key deterministically:
# run_id — unique per chain invocation (propagates into subgraphs)# event_type — on_tool_end / on_chain_end / on_llm_end# step_index — monotonic per-run counter you maintain in the handler
idempotency_key = f"{run_id}:{event_type}:{step_index}"
Retry budget: 1s → 5s → 30s (~36s total) then DLQ. Retry on 5xx / 429 /
network only — 4xx (except 429) goes straight to DLQ. DLQ is a Redis Stream
or S3 prefix keyed by YYYY/MM/DD/run_id/idempotency_key.json; alarm on depth
growth. See Idempotency and Retry for
HMAC verify, at-least-once vs at-most-once, and 24h de-dup window sizing.
Step 6 — Never dispatch from BackgroundTasks (P60)
FastAPI BackgroundTasks run after the response closes — exactly wrong for
per-event dispatch. Events must go out during the chain:
# WRONG — events fire all at once after the stream ends@app.post("/chat")asyncdefchat(req: Request, bg: BackgroundTasks):
bg.add_task(agent.ainvoke, {"messages": [...]}) # late + no streamingreturn {"status": "accepted"}
# RIGHT — handler fires during the chain, each on_tool_end dispatches immediately@app.post("/chat")asyncdefchat(req: ChatReq):
handler = EventDispatchHandler(sink=webhook_sink, run_id=req.run_id)
try:
result = await agent.ainvoke(
{"messages": req.messages},
config={"callbacks": [handler], "configurable": {"thread_id": req.thread_id}},
)
finally:
await handler.drain(timeout=5.0) # flush in-flight dispatchesreturn {"result": result}
drain() awaits in-flight asyncio.create_task() dispatches up to 5s so
events aren't lost when the pod scales down mid-request.
Output
Async BaseCallbackHandler subclass with asyncio.create_task() fire-and-forget
Callbacks wired via config["callbacks"] at invoke time (subgraph-safe)
Event-taxonomy filter: dispatch on_tool_end + named on_chain_end + on_llm_end
drain() on shutdown to flush in-flight dispatches
Error Handling
Error
Cause
Fix
Handler fires on outer agent but not subagent
Bound via with_config at definition (P28)
Pass via config["callbacks"] at invoke(...) time
Webhook analytics lags generation duration
BackgroundTasks fire post-response (P60)
Dispatch from the callback handler, never from BackgroundTasks
Browser / Kafka saturates on long generations
Forwarded astream_events(v2) raw (P47)
Filter events server-side; dispatch only on_tool_end/on_chain_end/on_llm_end
SSE stream hangs, no end event
Proxy buffering (P46)
Set X-Accel-Buffering: no — see langchain-langgraph-streaming
Event loop freezes on slow webhook
Sync POST in callback (P48)
Subclass AsyncCallbackHandler; use asyncio.create_task()
Duplicate events downstream
At-least-once dispatch + retry
Receiver dedupes on Idempotency-Key header with 24h cache
Orphan asyncio.create_task never runs
GC collected the task
Hold a strong reference in self._tasks and discard on completion
Events lost on pod shutdown
In-flight tasks cancelled
Call await handler.drain(timeout=5.0) in endpoint finally block
4xx webhook errors retrying 3x
Retry logic retrying everything
Retry only on 5xx / 429 / network; 4xx goes straight to DLQ
Signature verification fails on receiver
Body re-serialized with different key order
Sign the exact bytes you send; use sort_keys=True in json.dumps
Examples
Named subgraph dispatch with a LangGraph agent
Planner subagent runs search_docs → summarize; outer agent needs a webhook
on summarize completion. Full wiring in Subgraph Propagation.
Fan-out: webhook + Kafka + Redis Streams in one invocation
CompositeSink dispatches to multiple child sinks via
asyncio.gather(..., return_exceptions=True) — one sink's failure doesn't block
others. See Dispatch Targets.
HMAC-signed webhook receiver with de-dup
Verify X-Signature-256, SETNX Idempotency-Key against Redis with 24h TTL,
200 on both replay and first-seen. See Idempotency and Retry.
