Python LangGraph Middleware

The Python package is threadplane-middleware. It is the Python LangGraph twin of @threadplane/middleware/langgraph: it binds browser-declared client-tool stubs onto a chat model and routes client-tool-only turns to END.

Install

pip install threadplane-middleware

The package depends on langchain-core>=0.3.0 and langgraph>=0.3.0.

Install your model provider package separately, for example langchain-openai when using ChatOpenAI.

Bind tools per run

Call bind_client_tools() inside your agent node. The browser sends the tool catalog with each run, so the model-visible tool list is request-scoped.

from langchain_openai import ChatOpenAI
from threadplane.middleware.langgraph import bind_client_tools
 
SERVER_TOOLS = []  # your server-side LangChain tools
base_llm = ChatOpenAI(model="gpt-4o-mini")
 
def agent_node(state):
    llm = bind_client_tools(base_llm, SERVER_TOOLS, state)
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

The helper reads state["tools"] first and falls back to state["client_tools"]. It appends each client tool as an explicit OpenAI function-tool dict:

{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Read local weather",
        "parameters": {"type": "object"},
    },
}

Route after the agent

Use route_after_agent() from a LangGraph conditional edge. It returns the server tools node name when the last model message contains a server or unknown tool call. It returns __end__ when the turn has only client tool calls or no tool calls.

from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode
from threadplane.middleware.langgraph import route_after_agent
 
server_tool_names = [tool.name for tool in SERVER_TOOLS]
 
def router(state):
    return route_after_agent(state, server_tool_names)
 
graph = StateGraph(...)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(SERVER_TOOLS))
graph.add_conditional_edges("agent", router, {"tools": "tools", "__end__": END})

You can override the returned route labels:

route_after_agent(state, server_tool_names, tools_node="server_tools", end="done")

Helper surface

from threadplane.middleware.langgraph import (
    a2ui_client_capabilities,
    announce_subagent,
    bind_client_tools,
    client_tool_names,
    client_tool_specs,
    emit_custom_event,
    has_client_tool_call,
    has_server_tool_call,
    last_message,
    route_after_agent,
)

Helper	Purpose
`bind_client_tools(llm, server_tools, state)`	Bind server tools plus the run's client-tool stubs onto a model.
`client_tool_specs(state)`	Convert the run catalog into OpenAI function-tool dicts.
`client_tool_names(state)`	Return the set of client-declared tool names.
`has_client_tool_call(state)`	Check whether the last message calls a known client tool.
`has_server_tool_call(state, server_tool_names)`	Check whether the last message calls a server or unknown tool.
`route_after_agent(state, server_tool_names)`	Return the server tools node or the end label for a conditional edge.
`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:

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.

Warning: 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.

The Angular side of this is documented in the AG-UI Custom Events guide.

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.