AG-UI LangGraph Integration

0.0.33 verified Thu Apr 16 auth: no python

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.

pip install ag-ui-langgraph

Common errors

error 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`.

fix

Modify 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", "").

error 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`.

fix

This 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.

error 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.

fix

Implement 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.

error 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.

fix

Add 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. ↓

fix Monitor official GitHub for updates. Consider patching locally or downgrading `pydantic` if warnings are disruptive and do not indicate a critical issue. Keep `langchain-core` updated to v1.x or higher as recommended by `langsmith` security patches.

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. ↓

fix Implement a factory pattern where `add_langgraph_fastapi_endpoint` accepts a function that returns a *fresh* agent instance for each request, rather than a shared singleton. Alternatively, refactor the `LangGraphAgent` to pass `active_run` as a local variable or via `contextvars` for true thread-safety.

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. ↓

fix Apply a fallback when retrieving `tool_call_name`, using `tool_msg.name or event.get("name", "")` to ensure a valid string is always provided to `ToolCallStartEvent`.

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). ↓

fix Update `ag-ui-langgraph` and related `langchain` packages to use `@langchain/core ^1.1.0` or higher, which requires `langsmith >= 0.5.0`. This may require manually overriding package manager dependencies if the direct dependency is not updated.

Imports

LangGraphAgent

from ag_ui_langgraph import LangGraphAgent

add_langgraph_fastapi_endpoint

from ag_ui_langgraph import add_langgraph_fastapi_endpoint

Quickstart

This quickstart demonstrates how to expose a simple LangGraph agent as a FastAPI endpoint using `ag-ui-langgraph`. It defines a basic `StateGraph` with a single node that calls an OpenAI model, then integrates it into a FastAPI application, ready to serve AG-UI compatible requests.

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!" }] }'