AG-UI LangGraph Integration
ag-ui-langgraph is a Python library that provides a comprehensive integration of the Agent User Interaction (AG-UI) protocol with LangGraph. It enables standardized, event-driven communication between frontend applications and LangGraph-powered AI agents, facilitating real-time streaming interactions and state synchronization. The current version is 0.0.33 and it appears to have a relatively active release cadence with frequent updates to align with LangChain/LangGraph developments.
Common errors
-
pydantic_core.ValidationError: 1 validation error for ToolCallStartEvent tool_call_name. Input should be a valid string [type=string_type, input_value=None, input_type=NoneType]
cause Attempting to create a `ToolCallStartEvent` with a `tool_call_name` that is `None`, which can happen during agent-to-agent tool calls where `ToolMessage.name` defaults to `None`.fixModify the `on_tool_end` handler in `ag_ui_langgraph/agent.py` (or check for an official update) to use a fallback for `tool_call_name`, such as `tool_msg.name or event.get("name", "")`. -
TypeError: cannot pickle 'itertools.count' object
cause This error can occur when using `add_langgraph_fastapi_endpoint` in conjunction with certain `MCP (Managed Code Playground)` server tools. It seems to stem from a serialization issue within the `make_json_safe()` function in `ag_ui_langgraph/agent.py` when attempting to process non-serializable objects from `itertools`.fixThis is a known issue. Check for updates to `ag-ui-langgraph` that address serialization of complex objects from tools. If no update is available, you may need to manually sanitize tool outputs before they are processed by the AG-UI event streaming. -
Cannot send 'STEP_FINISHED' for step that was not started
cause This usually indicates a race condition or state corruption when `add_langgraph_fastapi_endpoint` uses a shared singleton `LangGraphAgent` instance across multiple concurrent requests. Request interleaving can lead to an `active_run` state being reset or modified by one request while another is still processing, causing an invalid state transition.fixImplement a per-request agent instance. Instead of `add_langgraph_fastapi_endpoint(app, graph, "/agent")`, adapt it to accept a factory function that creates a new `LangGraphAgent` for each incoming request, ensuring isolated state. -
No checkpointer set (LangGraph)
cause When using LangGraph with FastAPI, if your compiled graph does not have a checkpointer configured, LangGraph cannot persist or manage state across runs or retrieve history, leading to this error, especially in a stateful agent setup.fixAdd a checkpointer to your LangGraph `StateGraph`. For example, `graph.add_node("checkpoint", checkpoint_node).` Consult the LangGraph Persistence guide for details on how to add `SqliteSaver` or other checkpointer implementations.
Warnings
- gotcha The `ag-ui-langgraph` adapter has been observed to throw warnings related to deprecated functions from `Pydantic` and `LangGraph` versions. While not always breaking, this can create noise in logs.
- breaking Concurrent requests to an endpoint created with `add_langgraph_fastapi_endpoint` can lead to state corruption. This is due to `LangGraphAgent` storing per-request state in a shared `self.active_run` instance variable, leading to interleaving and corrupted state when multiple requests hit the same singleton agent instance.
- gotcha When one LangGraph agent calls another as a tool (agent-to-agent communication), the `ToolMessage.name` can be `None`, causing a `Pydantic validation error` in `ToolCallStartEvent` and crashing the Server-Sent Events (SSE) stream.
- breaking Older versions of `ag-ui-langgraph` might depend on `@langchain/core ^0.3.80`, which transitively pulls in `langsmith < 0.4.6`. This older `langsmith` version is affected by `CVE-2026-25528` (SSRF via tracing header injection).
Install
-
pip install ag-ui-langgraph
Imports
- LangGraphAgent
from ag_ui_langgraph import LangGraphAgent
- add_langgraph_fastapi_endpoint
from ag_ui_langgraph import add_langgraph_fastapi_endpoint
Quickstart
import os
from fastapi import FastAPI
from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from ag_ui_langgraph import add_langgraph_fastapi_endpoint
# Ensure OPENAI_API_KEY is set in your environment
os.environ['OPENAI_API_KEY'] = os.environ.get('OPENAI_API_KEY', 'your_openai_api_key_here')
# Define your LangGraph workflow
class AgentState(MessagesState):
pass
def call_model(state):
messages = state['messages']
model = ChatOpenAI(model="gpt-4o-mini") # Using a cost-effective model for quickstart
response = model.invoke(messages)
return {"messages": [response]}
workflow = StateGraph(AgentState)
workflow.add_node("oracle", call_model)
workflow.set_entry_point("oracle")
workflow.set_finish_point("oracle")
graph = workflow.compile()
# Create a FastAPI app
app = FastAPI()
# Integrate LangGraph with FastAPI using ag-ui-langgraph
# The endpoint will be available at /agent
add_langgraph_fastapi_endpoint(app, graph, "/agent")
# To run this:
# 1. Save as, e.g., main.py
# 2. pip install 'fastapi[all]' uvicorn ag-ui-langgraph langchain-openai langgraph
# 3. uvicorn main:app --reload --port 8000
# 4. Access http://localhost:8000/agent with a compatible AG-UI frontend
# or use curl:
# curl -X POST "http://localhost:8000/agent" \
# -H "Content-Type: application/json" \
# -d '{ "thread_id": "test_thread_123", "messages": [{ "role": "user", "content": "Hello!" }] }'