LangGraph Python SDK

Installation

pip install copilotkit

The CopilotKit Python SDK includes LangGraph and LangChain as dependencies. Core dependencies include:

langgraph>=0.3.25,<1.1.0
langchain>=0.3.0
fastapi>=0.115.0,<1.0.0
ag-ui-langgraph>=0.0.24

Core Classes

LangGraphAgent

The main class for integrating LangGraph agents with CopilotKit.

from copilotkit import LangGraphAgent

agent = LangGraphAgent(
    name="email_agent",
    description="This agent sends emails",
    graph=compiled_graph,
)

name

str

required

The name of the agent. Must consist of alphanumeric characters, underscores, and hyphens only.

graph

CompiledStateGraph

required

The compiled LangGraph graph to use with the agent.

description

str

Optional description of the agent. Used when CopilotKit dynamically routes requests to agents.

langgraph_config

RunnableConfig

Optional LangChain/LangGraph configuration to use with the agent.

copilotkit_config

CopilotKitConfig

Optional CopilotKit-specific configuration for advanced customization.

Example

from langgraph.graph import StateGraph, MessagesState
from langchain_openai import ChatOpenAI
from copilotkit import LangGraphAgent

# Define your graph
workflow = StateGraph(MessagesState)
model = ChatOpenAI(model="gpt-4")

def agent_node(state):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
workflow.set_finish_point("agent")

compiled_graph = workflow.compile()

# Create CopilotKit agent
agent = LangGraphAgent(
    name="my_agent",
    description="A helpful AI assistant",
    graph=compiled_graph
)

CopilotKitState

The state class that extends LangGraph’s MessagesState to include CopilotKit-specific properties.

from copilotkit import CopilotKitState
from langgraph.graph import StateGraph

class MyState(CopilotKitState):
    custom_field: str

workflow = StateGraph(MyState)

State Properties:

messages (List[BaseMessage]): The conversation messages
copilotkit (CopilotKitProperties): CopilotKit-specific state
- actions (List[Any]): Available actions from the frontend
- context (List[CopilotKitContextItem]): Context items from the frontend

CopilotKitConfig

Advanced configuration for customizing LangGraph integration behavior.

from typing import List
from langchain_core.messages import BaseMessage

def custom_merge_state(
    *,
    state: dict,
    messages: List[BaseMessage],
    actions: List[Any],
    agent_name: str
):
    # Custom state merging logic
    return {**state, "messages": messages}

def custom_convert_messages(messages: List[Message]):
    # Custom message conversion logic
    return converted_messages

config = {
    "merge_state": custom_merge_state,
    "convert_messages": custom_convert_messages
}

agent = LangGraphAgent(
    name="custom_agent",
    graph=graph,
    copilotkit_config=config
)

merge_state

Callable

Function to customize how CopilotKit merges the agent state.Parameters:

state (dict): Current state
messages (List[BaseMessage]): Incoming messages
actions (List[Any]): Available actions
agent_name (str): Name of the agent

convert_messages

Callable

Function to customize how CopilotKit converts its messages to LangChain messages.Parameters:

messages (List[Message]): CopilotKit messages to convert

Configuration Functions

copilotkit_customize_config

Customize the LangGraph configuration for use in CopilotKit.

from copilotkit.langgraph import copilotkit_customize_config

config = copilotkit_customize_config(
    config,
    emit_messages=False,
    emit_tool_calls=["SearchTool", "CalculatorTool"]
)

base_config

RunnableConfig

The LangChain/LangGraph configuration to customize. Pass None to create a new configuration.

emit_messages

bool

Configure how messages are emitted. By default, all messages are emitted. Pass False to disable.

emit_tool_calls

bool | str | List[str]

Configure how tool calls are emitted. By default, all tool calls are emitted. Pass False to disable, or a string/list of strings to emit only specific tool calls.

emit_intermediate_state

List[IntermediateStateConfig]

Emit tool calls as streaming LangGraph state.

Emit Intermediate State Example

config = copilotkit_customize_config(
    config,
    emit_intermediate_state=[
        {
            "state_key": "steps",
            "tool": "SearchTool",
            "tool_argument": "steps"
        },
    ]
)

Runtime Functions

copilotkit_exit

Signals CopilotKit to exit the current agent after the run completes.

from copilotkit.langgraph import copilotkit_exit

async def my_node(state, config):
    # Perform some work
    await copilotkit_exit(config)
    return state

config

RunnableConfig

required

The LangGraph configuration.

Returns: Awaitable[bool] - Always returns True.

Note

Calling copilotkit_exit() does not immediately stop the agent. It signals to CopilotKit to stop after the current run completes.

copilotkit_emit_state

Emits intermediate state to CopilotKit during long-running nodes.

from copilotkit.langgraph import copilotkit_emit_state

async def long_running_node(state, config):
    for i in range(10):
        await some_operation(i)
        await copilotkit_emit_state(config, {"progress": i})
    return state

config

RunnableConfig

required

The LangGraph configuration.

state

Any

required

The state to emit. Must be JSON serializable.

Returns: Awaitable[bool] - Always returns True.

copilotkit_emit_message

Manually emits a message to CopilotKit.

from copilotkit.langgraph import copilotkit_emit_message
from langchain_core.messages import AIMessage

async def my_node(state, config):
    message = "Step 1 of 10 complete"
    await copilotkit_emit_message(config, message)
    
    # Still need to return the message from the node
    return {"messages": [AIMessage(content=message)]}

config

RunnableConfig

required

The LangGraph configuration.

message

str

required

The message to emit.

Returns: Awaitable[bool] - Always returns True.

Important

You must still return the messages from the node. copilotkit_emit_message only sends a preview during execution.

copilotkit_emit_tool_call

Manually emits a tool call to CopilotKit.

from copilotkit.langgraph import copilotkit_emit_tool_call

async def my_node(state, config):
    await copilotkit_emit_tool_call(
        config,
        name="SearchTool",
        args={"query": "AI agents", "max_results": 5}
    )
    return state

config

RunnableConfig

required

The LangGraph configuration.

name

str

required

The name of the tool to emit.

args

Dict[str, Any]

required

The arguments to emit.

Returns: Awaitable[bool] - Always returns True.

copilotkit_interrupt

Creates an interrupt in the LangGraph execution to request user input.

from copilotkit.langgraph import copilotkit_interrupt

def my_node(state):
    # Ask user for confirmation
    answer, response = copilotkit_interrupt(
        message="Do you want to proceed with this action?"
    )
    
    if answer.lower() == "yes":
        # Continue with action
        pass
    
    return state

message

str

A text message to show the user. Either message or action must be provided.

action

str

The name of an action to execute. Either message or action must be provided.

args

Dict[str, Any]

Optional arguments for the action (used with action parameter).

Returns: Tuple[str, List[BaseMessage]] - The user’s response and the full message history.

Example with Action

answer, response = copilotkit_interrupt(
    action="confirmDelete",
    args={"item_id": "123"}
)

Message Conversion

copilotkit_messages_to_langchain

Converts CopilotKit messages to LangChain format.

from copilotkit.langgraph import copilotkit_messages_to_langchain

converter = copilotkit_messages_to_langchain(use_function_call=False)
langchain_messages = converter(copilotkit_messages)

use_function_call

bool

default:"False"

Whether to use function_call format (legacy) or tool_calls format (recommended).

Returns: Callable[[List[Message]], List[BaseMessage]] - A function that converts messages.

langchain_messages_to_copilotkit

Converts LangChain messages to CopilotKit format.

from copilotkit.langgraph import langchain_messages_to_copilotkit

copilotkit_messages = langchain_messages_to_copilotkit(langchain_messages)

messages

List[BaseMessage]

required

The LangChain messages to convert.

Returns: List[Message] - The converted CopilotKit messages.

FastAPI Integration

add_fastapi_endpoint

Add CopilotKit endpoints to a FastAPI application.

from fastapi import FastAPI
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit import CopilotKitRemoteEndpoint, LangGraphAgent

app = FastAPI()
sdk = CopilotKitRemoteEndpoint(agents=[agent])

add_fastapi_endpoint(app, sdk, "/copilotkit")

fastapi_app

FastAPI

required

The FastAPI application instance.

sdk

CopilotKitRemoteEndpoint

required

The CopilotKit SDK instance with configured agents.

prefix

str

required

The URL prefix for CopilotKit endpoints (e.g., “/copilotkit”).

use_thread_pool

bool

default:"False"

Deprecated. Whether to use a thread pool executor.

max_workers

int

default:"10"

Maximum number of worker threads (used with use_thread_pool).

Complete Example

Here’s a complete example showing a LangGraph agent with CopilotKit:

from fastapi import FastAPI
from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from copilotkit import (
    CopilotKitRemoteEndpoint,
    LangGraphAgent,
    CopilotKitState
)
from copilotkit.integrations.fastapi import add_fastapi_endpoint
from copilotkit.langgraph import (
    copilotkit_customize_config,
    copilotkit_emit_state
)

# Define state
class AgentState(CopilotKitState):
    result: str = ""

# Create graph
workflow = StateGraph(AgentState)
model = ChatOpenAI(model="gpt-4")

async def agent_node(state: AgentState, config):
    # Emit intermediate state
    await copilotkit_emit_state(config, {"status": "processing"})
    
    # Process messages
    response = model.invoke(state["messages"])
    
    return {
        "messages": [response],
        "result": response.content
    }

workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
workflow.set_finish_point("agent")

compiled_graph = workflow.compile()

# Create CopilotKit agent
agent = LangGraphAgent(
    name="research_agent",
    description="A research assistant that helps with information gathering",
    graph=compiled_graph
)

# Setup FastAPI
app = FastAPI()
sdk = CopilotKitRemoteEndpoint(agents=[agent])
add_fastapi_endpoint(app, sdk, "/copilotkit")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Type Hints

The SDK uses proper Python type hints throughout:

from typing import List, Dict, Any, Optional, Union
from langchain_core.messages import BaseMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph.state import CompiledStateGraph

# All functions and classes are fully typed
async def copilotkit_emit_state(
    config: RunnableConfig,
    state: Any
) -> bool:
    ...

Error Handling

The SDK includes custom exceptions for error handling:

from copilotkit.exc import (
    AgentNotFoundException,
    AgentExecutionException
)

try:
    result = await sdk.execute_agent(...)
except AgentNotFoundException as e:
    print(f"Agent not found: {e}")
except AgentExecutionException as e:
    print(f"Agent execution failed: {e}")

​Installation

​Core Classes

​LangGraphAgent

​Example

​CopilotKitState

​CopilotKitConfig

​Configuration Functions

​copilotkit_customize_config

​Emit Intermediate State Example

​Runtime Functions

​copilotkit_exit

Note

​copilotkit_emit_state

​copilotkit_emit_message

Important

​copilotkit_emit_tool_call

​copilotkit_interrupt

​Example with Action

​Message Conversion

​copilotkit_messages_to_langchain

​langchain_messages_to_copilotkit

​FastAPI Integration

​add_fastapi_endpoint

​Complete Example

​Type Hints

​Error Handling

Installation

Core Classes

LangGraphAgent

Example

CopilotKitState

CopilotKitConfig

Configuration Functions

copilotkit_customize_config

Emit Intermediate State Example

Runtime Functions

copilotkit_exit

copilotkit_emit_state

copilotkit_emit_message

copilotkit_emit_tool_call

copilotkit_interrupt

Example with Action

Message Conversion

copilotkit_messages_to_langchain

langchain_messages_to_copilotkit

FastAPI Integration

add_fastapi_endpoint

Complete Example

Type Hints

Error Handling