feat(middleware): emit_custom_event, the Python helper that survives ag-ui-langgraph

The ag-ui-langgraph bridge consumes the graph through astream_events, so the
only path from a node to the adapter's customEvents() signal is
adispatch_custom_event. A get_stream_writer() write with stream_mode="custom"
surfaces at most as a raw event and is silently dropped — the payload never
reaches the client, with no error anywhere.

Adds threadplane.middleware.langgraph.emit_custom_event, an async wrapper
around adispatch_custom_event that accepts the node's config when the caller
has it and otherwise relies on the ambient run context. Exported from the
package's __all__. The pytest drives a real one-node graph and asserts both
calls arrive as on_custom_event through astream_events.

Documents it in the package README, the Python LangGraph guide, and the AG-UI
Custom Events guide, whose backend snippet now shows the helper.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: blove

apps/website/content/docs/ag-ui/guides/custom-events.mdx

@@ -35,30 +35,34 @@ The adapter JSON-parses `value` when it arrives as a string, so consumers always
 
 ### The working path under ag-ui-langgraph
 
-The `ag-ui-langgraph` bridge consumes the graph through `astream_events`, and it forwards every `on_custom_event` it sees one-for-one as an AG-UI `CUSTOM` frame carrying the same name and payload. LangChain's `adispatch_custom_event` is what puts an `on_custom_event` on that stream, so it is the call a node (or a callback handler running inside one) makes to reach `customEvents`:
+The `ag-ui-langgraph` bridge consumes the graph through `astream_events`, and it forwards every `on_custom_event` it sees one-for-one as an AG-UI `CUSTOM` frame carrying the same name and payload. LangChain's `adispatch_custom_event` is what puts an `on_custom_event` on that stream, so it is the call a node (or a callback handler running inside one) makes to reach `customEvents`.
+
+The `threadplane-middleware` Python package wraps that call as `emit_custom_event`, which is the recommended way to make it:
 
 ```python
-from langchain_core.callbacks import adispatch_custom_event
 from langchain_core.runnables import RunnableConfig
+from threadplane.middleware.langgraph import emit_custom_event
 
 async def analysis_node(state: State, config: RunnableConfig) -> State:
     # Emit a partial result as the node runs
-    await adispatch_custom_event(
-        "analysis_progress", {"step": "scoring", "pct": 42}
+    await emit_custom_event(
+        "analysis_progress", {"step": "scoring", "pct": 42}, config=config
     )
 
     # ... do more work ...
 
-    await adispatch_custom_event(
-        "analysis_progress", {"step": "scoring", "pct": 100}
+    await emit_custom_event(
+        "analysis_progress", {"step": "scoring", "pct": 100}, config=config
     )
     return state
 ```
 
+The signature is `emit_custom_event(name, value, *, config=None)`. Pass `config` when the node already receives one; omit it and the ambient run context is used. Backends that do not depend on the middleware package can call `adispatch_custom_event` from `langchain_core.callbacks` directly — the helper adds no wire behavior of its own.
+
 The event name becomes `CustomStreamEvent.name` and the payload becomes `CustomStreamEvent.data`. This is the mechanism the [subagents example](/docs/ag-ui/guides/subagents) uses to stream child-agent tokens from a callback handler.
 
 <Callout type="warning" title="get_stream_writer does not survive ag-ui-langgraph">
-Writing to `get_stream_writer()` with `stream_mode='custom'` does **not** produce a `CUSTOM` frame under `ag-ui-langgraph`. The bridge reads `astream_events`, where a stream-writer write surfaces at most as a raw event, so nothing is appended to `customEvents`. Use `adispatch_custom_event` instead. Other AG-UI runtimes that emit `CUSTOM` frames directly are unaffected by this constraint — the adapter only cares that a `CUSTOM` frame arrives.
+Writing to `get_stream_writer()` with `stream_mode='custom'` does **not** produce a `CUSTOM` frame under `ag-ui-langgraph`. The bridge reads `astream_events`, where a stream-writer write surfaces at most as a raw event, so nothing is appended to `customEvents`. Use `emit_custom_event` (or `adispatch_custom_event`) instead. Other AG-UI runtimes that emit `CUSTOM` frames directly are unaffected by this constraint — the adapter only cares that a `CUSTOM` frame arrives.
 </Callout>
 
 ### Graph state is a different signal

apps/website/content/docs/middleware/guides/python-langgraph.mdx

@@ -82,6 +82,7 @@ from threadplane.middleware.langgraph import (
     bind_client_tools,
     client_tool_names,
     client_tool_specs,
+    emit_custom_event,
     has_client_tool_call,
     has_server_tool_call,
     last_message,
@@ -100,9 +101,31 @@ from threadplane.middleware.langgraph import (
 | `last_message(state)` | Return the last message from `state["messages"]`, or `None`. |
 | `a2ui_client_capabilities(state)` | Return the A2UI capabilities the frontend advertised, or `None`. |
 | `announce_subagent(config, tool_call_id)` | Emit a custom event binding a child graph's stream namespace to the tool call that started it. |
+| `emit_custom_event(name, value, config=None)` | Push a payload to the frontend as an AG-UI `CUSTOM` event. |
 
 That import list is the package's full `__all__`.
 
+## Pushing data to the frontend mid-run
+
+`emit_custom_event` is an async helper that wraps LangChain's `adispatch_custom_event`:
+
+```python
+from langchain_core.runnables import RunnableConfig
+from threadplane.middleware.langgraph import emit_custom_event
+
+async def analysis_node(state: State, config: RunnableConfig) -> State:
+    await emit_custom_event("analysis_progress", {"pct": 42}, config=config)
+    return state
+```
+
+The `name` becomes `CustomStreamEvent.name` on the client and the `value` becomes `CustomStreamEvent.data`. Pass `config` when the node already receives one; omit it and the ambient run context is used.
+
+<Callout type="warning" title="get_stream_writer does not reach the frontend">
+An `ag-ui-langgraph` backend consumes the graph through `astream_events`, and only `adispatch_custom_event` places an event on that stream. Writing to `get_stream_writer()` with `stream_mode="custom"` is silently dropped, so nothing reaches the adapter. Use `emit_custom_event` and the payload survives.
+</Callout>
+
+The Angular side of this is documented in the AG-UI [Custom Events guide](/docs/ag-ui/guides/custom-events).
+
 ## Frontend contract
 
 The middleware does not execute browser tools. The frontend still needs to send the catalog, observe the model tool call, execute the local function or UI interaction, and resume the graph with a `ToolMessage` containing the result.

packages/threadplane-middleware/README.md

@@ -71,6 +71,24 @@ from threadplane.middleware.langgraph import (
 )
 ```
 
+## Pushing data to the frontend mid-run
+
+```python
+from langchain_core.runnables import RunnableConfig
+from threadplane.middleware.langgraph import emit_custom_event
+
+async def analysis_node(state, config: RunnableConfig):
+    await emit_custom_event("analysis_progress", {"pct": 42}, config=config)
+    return state
+```
+
+`emit_custom_event(name, value, *, config=None)` wraps LangChain's
+`adispatch_custom_event`. An `ag-ui-langgraph` backend consumes the graph
+through `astream_events`, and only `adispatch_custom_event` places an event on
+that stream — a `get_stream_writer()` write with `stream_mode="custom"` is
+silently dropped and never reaches the client. Pass `config` when the node
+already receives one; omit it and the ambient run context is used.
+
 ## Development
 
 ```bash

packages/threadplane-middleware/src/threadplane/middleware/langgraph/__init__.py

@@ -1,5 +1,6 @@
 """threadplane-middleware — LangGraph middleware for client-declared tools."""
 
+from threadplane.middleware.langgraph.custom_events import emit_custom_event
 from threadplane.middleware.langgraph.middleware import (
     a2ui_client_capabilities,
     announce_subagent,
@@ -18,6 +19,7 @@
     "bind_client_tools",
     "client_tool_names",
     "client_tool_specs",
+    "emit_custom_event",
     "has_client_tool_call",
     "has_server_tool_call",
     "last_message",

packages/threadplane-middleware/src/threadplane/middleware/langgraph/custom_events.py

@@ -0,0 +1,43 @@
+"""Emit a custom event that survives the ag-ui-langgraph bridge."""
+
+from typing import Any, Optional
+
+from langchain_core.callbacks.manager import adispatch_custom_event
+
+
+async def emit_custom_event(
+    name: str,
+    value: Any,
+    *,
+    config: Optional[Any] = None,
+) -> None:
+    """Push ``value`` to the frontend as an AG-UI ``CUSTOM`` event.
+
+    The ``ag-ui-langgraph`` bridge consumes the graph through ``astream_events``
+    and forwards every ``on_custom_event`` it sees as a ``CUSTOM`` frame. Only
+    ``adispatch_custom_event`` puts an ``on_custom_event`` on that stream:
+    writing to ``get_stream_writer()`` with ``stream_mode="custom"`` surfaces at
+    most as a raw event and is silently dropped, so nothing reaches the
+    adapter's ``customEvents()`` signal. This helper is that call, named for
+    what it does::
+
+        from langchain_core.runnables import RunnableConfig
+        from threadplane.middleware.langgraph import emit_custom_event
+
+        async def analysis_node(state: State, config: RunnableConfig) -> State:
+            await emit_custom_event("analysis_progress", {"pct": 42}, config=config)
+            return state
+
+    ``name`` becomes ``CustomStreamEvent.name`` on the client and ``value``
+    becomes ``CustomStreamEvent.data``.
+
+    Pass ``config`` when the node already receives one — LangChain then dispatches
+    through that config's callback manager rather than the ambient contextvar,
+    which is what keeps the event attributed correctly inside nested runnables
+    and on Python 3.10, where the contextvar is not propagated automatically.
+    Omit it and the ambient run context is used.
+    """
+    if config is None:
+        await adispatch_custom_event(name, value)
+    else:
+        await adispatch_custom_event(name, value, config=config)

packages/threadplane-middleware/tests/test_custom_events.py

@@ -0,0 +1,51 @@
+"""Tests for threadplane.middleware.langgraph.custom_events.
+
+``emit_custom_event`` must reach ``astream_events`` as an ``on_custom_event``,
+because that stream is the only path an ``ag-ui-langgraph`` bridge reads. A
+``get_stream_writer`` write does not surface there, so the helper wraps
+``adispatch_custom_event``.
+"""
+
+import asyncio
+
+from langgraph.graph import END, StateGraph
+from typing_extensions import TypedDict
+
+from threadplane.middleware.langgraph import emit_custom_event
+
+
+class _State(TypedDict, total=False):
+    value: int
+
+
+def _build_graph():
+    async def node(state: _State, config=None) -> _State:
+        # Once with the node's config, once relying on the contextvar.
+        await emit_custom_event("analysis_progress", {"pct": 42}, config=config)
+        await emit_custom_event("analysis_progress", {"pct": 100})
+        return {"value": 1}
+
+    graph = StateGraph(_State)
+    graph.add_node("work", node)
+    graph.add_edge("__start__", "work")
+    graph.add_edge("work", END)
+    return graph.compile()
+
+
+def _collect_custom_events():
+    async def run():
+        graph = _build_graph()
+        seen = []
+        async for event in graph.astream_events({"value": 0}, version="v2"):
+            if event["event"] == "on_custom_event":
+                seen.append((event["name"], event["data"]))
+        return seen
+
+    return asyncio.run(run())
+
+
+def test_emit_custom_event_reaches_astream_events():
+    assert _collect_custom_events() == [
+        ("analysis_progress", {"pct": 42}),
+        ("analysis_progress", {"pct": 100}),
+    ]