Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
Designed for Claude Code, also compatible with Codex
LangChain Webhooks and Event Dispatch (Python)
Overview
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.
