diff --git a/apps/website/content/docs/ag-ui/guides/custom-events.mdx b/apps/website/content/docs/ag-ui/guides/custom-events.mdx index 9e4987223..64be1ac8b 100644 --- a/apps/website/content/docs/ag-ui/guides/custom-events.mdx +++ b/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. -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. ### Graph state is a different signal diff --git a/apps/website/content/docs/middleware/api/api-docs.json b/apps/website/content/docs/middleware/api/api-docs.json index 3df551fab..7b0f3baf9 100644 --- a/apps/website/content/docs/middleware/api/api-docs.json +++ b/apps/website/content/docs/middleware/api/api-docs.json @@ -559,7 +559,7 @@ { "name": "clientToolsRouter", "kind": "function", - "description": "A prebuilt conditional-edge callback. serverToolNames is bound once at construction;\nthe returned function takes only state.\n\n graph.addConditionalEdges('agent', clientToolsRouter([]), ['tools', END]);", + "description": "A prebuilt conditional-edge callback. serverToolNames is bound once at construction;\nthe returned function takes only state.\n\n graph.addConditionalEdges('agent', clientToolsRouter(names), ['server_tools', END]);\n\n`opts.toolsNode` defaults to `'server_tools'`; a graph carrying the client-tool\nchannels cannot name a node `tools`, because clientToolsChannel already\nclaims that name as a state channel.", "signature": "clientToolsRouter(serverToolNames: Iterable, opts: object): (state: ClientToolsState) => string", "params": [ { @@ -760,7 +760,7 @@ { "name": "routeAfterAgent", "kind": "function", - "description": "Routing helper for a LangGraph conditional edge. Returns `toolsNode` when the last\nmessage has a server tool call (dispatch to the server ToolNode); otherwise `end`\n(client-only calls — the browser executes them — and no-tool-call turns both end).", + "description": "Routing helper for a LangGraph conditional edge. Returns `toolsNode` when the last\nmessage has a server tool call (dispatch to the server ToolNode); otherwise `end`\n(client-only calls — the browser executes them — and no-tool-call turns both end).\n\n`toolsNode` defaults to `'server_tools'`. It cannot default to `'tools'`, because\nclientToolsChannel declares a `tools` state channel and LangGraph.js shares\none namespace between channel names and node names — `addNode('tools', …)` throws\n\"tools is already being used as a state attribute\".", "signature": "routeAfterAgent(state: ClientToolsState, serverToolNames: Iterable, opts: object): string", "params": [ { diff --git a/apps/website/content/docs/middleware/getting-started/introduction.mdx b/apps/website/content/docs/middleware/getting-started/introduction.mdx index ecfdb6889..e43b5b1d5 100644 --- a/apps/website/content/docs/middleware/getting-started/introduction.mdx +++ b/apps/website/content/docs/middleware/getting-started/introduction.mdx @@ -54,7 +54,7 @@ If a turn mixes server tool calls and client tool calls, server tools win the fi |-----|---------| | `clientToolsChannel()` | Adds the `tools` and `client_tools` state channels to a LangGraph annotation. | | `bindClientTools()` | Binds server tools plus client-declared tool stubs onto a model. | -| `clientToolsRouter()` | Creates a conditional-edge router for server-tool vs client-tool routing. | +| `clientToolsRouter()` | Creates a conditional-edge router for server-tool vs client-tool routing. Its server destination defaults to `'server_tools'`, because `tools` is already a state channel and LangGraph.js forbids a node of that name. | | `clientToolSpecs()` | Converts state catalog entries into OpenAI function-tool specs. | | `clientToolNames()` | Returns the set of client-declared tool names for a run. | | `hasClientToolCall()` | Checks whether the last message calls a client tool. | @@ -89,6 +89,7 @@ The same entry point also exports a deduplication surface, for backends that mus | `last_message()` | Reads the last message from state. | | `a2ui_client_capabilities(state)` | Reads the A2UI client capabilities the frontend advertised, or `None` when it advertised none. | | `announce_subagent(config, tool_call_id)` | Emits a custom event binding a child graph's stream namespace to the tool call that started it. | +| `emit_custom_event(name, value, config=None)` | Pushes a payload to the frontend as an AG-UI `CUSTOM` event, on the one delivery path an `ag-ui-langgraph` bridge reads. | ## When to use it diff --git a/apps/website/content/docs/middleware/getting-started/quickstart.mdx b/apps/website/content/docs/middleware/getting-started/quickstart.mdx index fa0bad56f..de7895fd6 100644 --- a/apps/website/content/docs/middleware/getting-started/quickstart.mdx +++ b/apps/website/content/docs/middleware/getting-started/quickstart.mdx @@ -65,19 +65,19 @@ const graph = new StateGraph(State) .addEdge('server_tools', 'agent') .addConditionalEdges( 'agent', - (state) => clientToolsRouter(serverToolNames, { toolsNode: 'server_tools' })(state), + (state) => clientToolsRouter(serverToolNames)(state), ['server_tools', END], ) .compile(); ``` -When the last model message calls a server tool, the router returns `'server_tools'`. When the last model message calls only browser-declared client tools, the router returns `END` so the frontend can execute the call and resume. +When the last model message calls a server tool, the router returns `'server_tools'` — its default destination, which is why the snippet above passes no options. When the last model message calls only browser-declared client tools, the router returns `END` so the frontend can execute the call and resume. - -`clientToolsChannel()` declares a `tools` state channel, and LangGraph refuses a node whose name collides with a channel: `addNode` throws *"tools is already being used as a state attribute (a.k.a. a channel), cannot also be used as a node name"*. The router's default destination is `'tools'`, so pass `toolsNode` whenever the graph actually has a server tool node. Every destination named in the path map must also exist as a node, or `.compile()` throws *"Found edge ending at unknown node"*. + +`clientToolsChannel()` declares a `tools` state channel, and LangGraph refuses a node whose name collides with a channel: `addNode` throws *"tools is already being used as a state attribute (a.k.a. a channel), cannot also be used as a node name"*. That is why the router's default destination is `'server_tools'`. Name the node something else and pass `toolsNode` if `'server_tools'` does not suit you. Every destination named in the path map must also exist as a node, or `.compile()` throws *"Found edge ending at unknown node"*. -A graph with no server tools at all can drop the node, the override, and the path-map entry, and route to `[END]` alone — which is what the package's own integration test does. +A graph with no server tools at all can drop the node and the path-map entry, and route to `[END]` alone — which is what the package's own integration test does. ## Complete skeleton @@ -114,7 +114,7 @@ export const graph = new StateGraph(State) .addEdge('server_tools', 'agent') .addConditionalEdges( 'agent', - (state) => clientToolsRouter(serverToolNames, { toolsNode: 'server_tools' })(state), + (state) => clientToolsRouter(serverToolNames)(state), ['server_tools', END], ) .compile(); diff --git a/apps/website/content/docs/middleware/guides/langgraph-client-tools.mdx b/apps/website/content/docs/middleware/guides/langgraph-client-tools.mdx index 2a6a7ba46..6c5909bd9 100644 --- a/apps/website/content/docs/middleware/guides/langgraph-client-tools.mdx +++ b/apps/website/content/docs/middleware/guides/langgraph-client-tools.mdx @@ -61,14 +61,14 @@ The router returns `END` when the last model message has only known client-tool ```ts .addConditionalEdges( 'agent', - (state) => clientToolsRouter(['lookupOrder'], { toolsNode: 'server_tools' })(state), + (state) => clientToolsRouter(['lookupOrder'])(state), ['server_tools', END], ) ``` Use `serverToolNames` to disambiguate tools you actually execute on the backend. -Two rules govern the destination name. Every node named in the path map must already exist on the graph, or `.compile()` throws *"Found edge ending at unknown node"*. And the node cannot be called `tools`, because `clientToolsChannel()` declares a `tools` state channel and LangGraph rejects a node name that collides with a channel — hence the `toolsNode` override above. +Two rules govern the destination name. Every node named in the path map must already exist on the graph, or `.compile()` throws *"Found edge ending at unknown node"*. And the node cannot be called `tools`, because `clientToolsChannel()` declares a `tools` state channel and LangGraph rejects a node name that collides with a channel. The router's default destination is therefore `'server_tools'`; pass `{ toolsNode }` only when your server tool node carries a different name. ## Mixed tool calls @@ -91,12 +91,12 @@ import { } from '@threadplane/middleware/langgraph'; ``` -`routeAfterAgent(state, serverToolNames, opts)` is the primitive behind `clientToolsRouter()`. The default destinations are `'tools'` and `'__end__'`. Because `'tools'` is unusable as a node name on a graph that carries the client-tool channels, override it whenever a server tool node exists: +`routeAfterAgent(state, serverToolNames, opts)` is the primitive behind `clientToolsRouter()`. The default destinations are `'server_tools'` and `'__end__'`. Override either one when your graph names those nodes differently: ```ts routeAfterAgent(state, ['lookupOrder'], { - toolsNode: 'server_tools', - end: '__end__', + toolsNode: 'backend_tools', + end: 'wrap_up', }); ``` diff --git a/apps/website/content/docs/middleware/guides/python-langgraph.mdx b/apps/website/content/docs/middleware/guides/python-langgraph.mdx index 82a6cae42..47272b2c4 100644 --- a/apps/website/content/docs/middleware/guides/python-langgraph.mdx +++ b/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. + + +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. + + +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. diff --git a/libs/middleware/CHANGELOG.md b/libs/middleware/CHANGELOG.md new file mode 100644 index 000000000..67f51dd18 --- /dev/null +++ b/libs/middleware/CHANGELOG.md @@ -0,0 +1,9 @@ +# Changelog + +## [Unreleased] + +### Breaking + +- `routeAfterAgent()` and `clientToolsRouter()` now default `toolsNode` to `'server_tools'` instead of `'tools'`. The old default could never work: `clientToolsChannel()` declares a `tools` state channel, and LangGraph.js shares one namespace between channel names and node names, so `addNode('tools', …)` throws *"tools is already being used as a state attribute"* on exactly the graphs these helpers are for. Any graph that relied on the old default was already passing `{ toolsNode: 'server_tools' }` (or an equivalent override) to work around it. Rename your server tool node to `server_tools` and drop the override, or keep the override pointing at whatever name your node uses. There is no shim. + + The Python package's `route_after_agent()` keeps `tools_node="tools"`; Python LangGraph does not share that namespace. diff --git a/libs/middleware/README.md b/libs/middleware/README.md index 30e93cc86..f236683eb 100644 --- a/libs/middleware/README.md +++ b/libs/middleware/README.md @@ -55,10 +55,17 @@ const graph = new StateGraph(State) .addNode('agent', agent) .addEdge('__start__', 'agent') // clientToolsRouter binds the server tool names once; pass [] when there are none. - .addConditionalEdges('agent', clientToolsRouter([]), ['tools', END]) + // With no server tools there is no tool node, so END is the only destination. + .addConditionalEdges('agent', clientToolsRouter([]), [END]) .compile(); ``` +The router's server destination defaults to `'server_tools'`. It cannot default to +`'tools'`: `clientToolsChannel()` declares a `tools` state channel, and LangGraph.js +shares one namespace between channel names and node names, so `addNode('tools', …)` +throws *"tools is already being used as a state attribute"*. Name the server tool node +`server_tools` (or pass `{ toolsNode }` to use another name). + ### What happens with a client tool call 1. The model emits a tool call whose name matches a client-declared tool. diff --git a/libs/middleware/src/integration.spec.ts b/libs/middleware/src/integration.spec.ts index 9c9035daf..8832b5153 100644 --- a/libs/middleware/src/integration.spec.ts +++ b/libs/middleware/src/integration.spec.ts @@ -1,6 +1,9 @@ import { describe, it, expect } from 'vitest'; import { Annotation, MessagesAnnotation, StateGraph, END } from '@langchain/langgraph'; +import { ToolNode } from '@langchain/langgraph/prebuilt'; import { AIMessage, ToolMessage, HumanMessage } from '@langchain/core/messages'; +import { tool } from '@langchain/core/tools'; +import { z } from 'zod'; import { bindClientTools, clientToolsChannel, clientToolsRouter } from './langgraph'; // A scripted fake chat model exposing the bindTools + invoke surface the graph uses. @@ -51,3 +54,43 @@ describe('client-tools loop (in-process)', () => { expect((r2.messages[r2.messages.length - 1] as AIMessage).content).toBe('It is 65F in SF.'); }); }); + +describe("the router's default toolsNode", () => { + it('dispatches a server tool call to a node named by the default, with no override', async () => { + const echo = tool(async ({ text }: { text: string }) => `echoed:${text}`, { + name: 'echo', + description: 'Echo the input.', + schema: z.object({ text: z.string() }), + }); + + let agentTurns = 0; + const graph = new StateGraph(State) + .addNode('agent', async () => { + agentTurns += 1; + if (agentTurns > 1) return { messages: [new AIMessage({ content: 'done' })] }; + return { + messages: [ + new AIMessage({ content: '', tool_calls: [{ name: 'echo', args: { text: 'hi' }, id: 'call_1' }] }), + ], + }; + }) + .addNode('server_tools', new ToolNode([echo])) + .addEdge('__start__', 'agent') + .addEdge('server_tools', 'agent') + .addConditionalEdges('agent', (s) => clientToolsRouter(['echo'])(s), ['server_tools', END]) + .compile(); + + const result = await graph.invoke({ messages: [new HumanMessage('echo hi')] }); + const toolMessage = result.messages.find((m): m is ToolMessage => m instanceof ToolMessage); + expect(toolMessage?.content).toBe('echoed:hi'); + }); + + it("records why 'tools' cannot be a node name on a client-tools graph", () => { + // clientToolsChannel() declares a `tools` state channel, and LangGraph JS + // shares one namespace between channel names and node names — which is why + // the router's default destination is 'server_tools', not 'tools'. + expect(() => new StateGraph(State).addNode('tools', async () => ({ messages: [] }))).toThrow( + /tools is already being used as a state attribute/, + ); + }); +}); diff --git a/libs/middleware/src/langgraph.spec.ts b/libs/middleware/src/langgraph.spec.ts index 30592e4d1..4d0f09c4b 100644 --- a/libs/middleware/src/langgraph.spec.ts +++ b/libs/middleware/src/langgraph.spec.ts @@ -105,7 +105,7 @@ describe('routeAfterAgent', () => { tools: [{ name: 'get_weather', description: '', parameters: {} }], }); it('routes a server tool call to the tools node', () => { - expect(routeAfterAgent(st(['search']), ['search'])).toBe('tools'); + expect(routeAfterAgent(st(['search']), ['search'])).toBe('server_tools'); }); it('routes a client-only tool call to END', () => { expect(routeAfterAgent(st(['get_weather']), [])).toBe('__end__'); @@ -114,7 +114,7 @@ describe('routeAfterAgent', () => { expect(routeAfterAgent(st([]), [])).toBe('__end__'); }); it('routes a mixed call to the server (precedence)', () => { - expect(routeAfterAgent(st(['get_weather', 'search']), ['search'])).toBe('tools'); + expect(routeAfterAgent(st(['get_weather', 'search']), ['search'])).toBe('server_tools'); }); it('honors custom node names', () => { expect(routeAfterAgent(st(['search']), ['search'], { toolsNode: 'act' })).toBe('act'); @@ -131,7 +131,7 @@ describe('clientToolsRouter', () => { }); it('returns a callback that routes via routeAfterAgent with bound serverToolNames', () => { const route = clientToolsRouter(['search']); - expect(route(st(['search']))).toBe('tools'); + expect(route(st(['search']))).toBe('server_tools'); expect(route(st(['get_weather']))).toBe('__end__'); }); it('honors custom node names', () => { diff --git a/libs/middleware/src/langgraph/middleware.ts b/libs/middleware/src/langgraph/middleware.ts index 3111f25cc..c4a600463 100644 --- a/libs/middleware/src/langgraph/middleware.ts +++ b/libs/middleware/src/langgraph/middleware.ts @@ -85,13 +85,18 @@ export function bindClientTools( * Routing helper for a LangGraph conditional edge. Returns `toolsNode` when the last * message has a server tool call (dispatch to the server ToolNode); otherwise `end` * (client-only calls — the browser executes them — and no-tool-call turns both end). + * + * `toolsNode` defaults to `'server_tools'`. It cannot default to `'tools'`, because + * {@link clientToolsChannel} declares a `tools` state channel and LangGraph.js shares + * one namespace between channel names and node names — `addNode('tools', …)` throws + * "tools is already being used as a state attribute". */ export function routeAfterAgent( state: ClientToolsState, serverToolNames: Iterable, opts?: { toolsNode?: string; end?: string }, ): string { - const toolsNode = opts?.toolsNode ?? 'tools'; + const toolsNode = opts?.toolsNode ?? 'server_tools'; const end = opts?.end ?? '__end__'; return hasServerToolCall(state, serverToolNames) ? toolsNode : end; } diff --git a/libs/middleware/src/langgraph/router.ts b/libs/middleware/src/langgraph/router.ts index fcbed9a7f..e25c5add5 100644 --- a/libs/middleware/src/langgraph/router.ts +++ b/libs/middleware/src/langgraph/router.ts @@ -5,7 +5,11 @@ import type { ClientToolsState } from './types.js'; * A prebuilt conditional-edge callback. serverToolNames is bound once at construction; * the returned function takes only state. * - * graph.addConditionalEdges('agent', clientToolsRouter([]), ['tools', END]); + * graph.addConditionalEdges('agent', clientToolsRouter(names), ['server_tools', END]); + * + * `opts.toolsNode` defaults to `'server_tools'`; a graph carrying the client-tool + * channels cannot name a node `tools`, because {@link clientToolsChannel} already + * claims that name as a state channel. */ export function clientToolsRouter( serverToolNames: Iterable, diff --git a/packages/threadplane-middleware/README.md b/packages/threadplane-middleware/README.md index 9cffdbcd3..6b9e00948 100644 --- a/packages/threadplane-middleware/README.md +++ b/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 diff --git a/packages/threadplane-middleware/src/threadplane/middleware/langgraph/__init__.py b/packages/threadplane-middleware/src/threadplane/middleware/langgraph/__init__.py index f352bac29..524bc13a7 100644 --- a/packages/threadplane-middleware/src/threadplane/middleware/langgraph/__init__.py +++ b/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", diff --git a/packages/threadplane-middleware/src/threadplane/middleware/langgraph/custom_events.py b/packages/threadplane-middleware/src/threadplane/middleware/langgraph/custom_events.py new file mode 100644 index 000000000..47da9dadd --- /dev/null +++ b/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) diff --git a/packages/threadplane-middleware/tests/test_custom_events.py b/packages/threadplane-middleware/tests/test_custom_events.py new file mode 100644 index 000000000..2f2dca9aa --- /dev/null +++ b/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}), + ]