[BUG] langgraph: AttributeError: 'Command' object has no attribute 'tool_call_id' when tool returns Command object

[EcoleKeine]

Error when using langgraph_supervisor create_supervisor()

Log

  File "...\.venv\Lib\site-packages\ag_ui_langgraph\agent.py", line 106, in run
    async for event_str in self._handle_stream_events(input.copy(update={"forwarded_props": forwarded_props})):
        yield event_str
  File "...\.venv\Lib\site-packages\ag_ui_langgraph\agent.py", line 211, in _handle_stream_events
    async for single_event in self._handle_single_event(event, state):
        yield single_event
  File "...\.venv\Lib\site-packages\copilotkit\langgraph_agui_agent.py", line 167, in _handle_single_event
    async for event_str in super()._handle_single_event(event, state):
        yield event_str
  File "...\.venv\Lib\site-packages\ag_ui_langgraph\agent.py", line 677, in _handle_single_event
    tool_call_id=tool_call_output.tool_call_id,
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'Command' object has no attribute 'tool_call_id'
WARNING:langgraph:Task supervisor with path ('__pregel_pull', 'supervisor') wrote to unknown channel remaining_steps, ignoring it.
ERROR:asyncio:Task exception was never retrieved

According to the LangGraph documentation https://langchain-ai.github.io/langgraph/how-tos/tool-calling/#short-term-memory, tools can return a langgraph.types.Command object

[7AM7]

I faced the same issue when use
tool.invoke() it fails
You should use
tool.run()
No idea way but this works well for me

[johandieman]

I wrote a monkey patch for this seems like the langgraph Command events aren't directly supported. In my case it is omitted as a subgraph routing command.

import json
import uuid
from typing import Any, AsyncGenerator
from langchain_core.messages import ToolMessage
from langgraph.types import Command


async def patched_handle_single_event(self, event: Any, state) -> AsyncGenerator[str, None]:
    from ag_ui_langgraph.agent import (
        LangGraphEventTypes,
        CustomEventNames,
        EventType,
        MessageInProgress,
        TextMessageStartEvent,
        TextMessageContentEvent,
        TextMessageEndEvent,
        ToolCallStartEvent,
        ToolCallArgsEvent,
        ToolCallEndEvent,
        ToolCallResultEvent,
        CustomEvent,
        StateSnapshotEvent,
        ThinkingTextMessageEndEvent,
        ThinkingEndEvent,
        resolve_reasoning_content,
        resolve_message_content,
    )

    event_type = event.get("event")

    if event_type == LangGraphEventTypes.OnChatModelStream:
        should_emit_messages = event["metadata"].get("emit-messages", True)
        should_emit_tool_calls = event["metadata"].get("emit-tool-calls", True)

        if event["data"]["chunk"].response_metadata.get('finish_reason', None):
            return

        current_stream = self.get_message_in_progress(self.active_run["id"])
        has_current_stream = bool(current_stream and current_stream.get("id"))
        tool_call_data = event["data"]["chunk"].tool_call_chunks[0] if event["data"]["chunk"].tool_call_chunks else None
        predict_state_metadata = event["metadata"].get("predict_state", [])
        tool_call_used_to_predict_state = False
        if tool_call_data and tool_call_data.get("name") and predict_state_metadata:
            tool_call_used_to_predict_state = any(
                predict_tool.get("tool") == tool_call_data["name"]
                for predict_tool in predict_state_metadata
            )

        is_tool_call_start_event = not has_current_stream and tool_call_data and tool_call_data.get("name")
        is_tool_call_args_event = has_current_stream and current_stream.get(
            "tool_call_id") and tool_call_data and tool_call_data.get("args")
        is_tool_call_end_event = has_current_stream and current_stream.get("tool_call_id") and not tool_call_data

        if is_tool_call_start_event or is_tool_call_end_event or is_tool_call_args_event:
            self.active_run["has_function_streaming"] = True

        reasoning_data = resolve_reasoning_content(event["data"]["chunk"]) if event["data"]["chunk"] else None
        message_content = resolve_message_content(event["data"]["chunk"].content) if event["data"]["chunk"] and \
                                                                                     event["data"][
                                                                                         "chunk"].content else None
        is_message_content_event = tool_call_data is None and message_content
        is_message_end_event = has_current_stream and not current_stream.get(
            "tool_call_id") and not is_message_content_event

        if reasoning_data:
            self.handle_thinking_event(reasoning_data)
            return

        if reasoning_data is None and self.active_run.get('thinking_process', None) is not None:
            yield self._dispatch_event(
                ThinkingTextMessageEndEvent(
                    type=EventType.THINKING_TEXT_MESSAGE_END,
                )
            )
            yield self._dispatch_event(
                ThinkingEndEvent(
                    type=EventType.THINKING_END,
                )
            )
            self.active_run["thinking_process"] = None

        if tool_call_used_to_predict_state:
            yield self._dispatch_event(
                CustomEvent(
                    type=EventType.CUSTOM,
                    name="PredictState",
                    value=predict_state_metadata,
                    raw_event=event
                )
            )

        if is_tool_call_end_event:
            yield self._dispatch_event(
                ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=current_stream["tool_call_id"],
                                 raw_event=event)
            )
            self.messages_in_process[self.active_run["id"]] = None
            return

        if is_message_end_event:
            yield self._dispatch_event(
                TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END, message_id=current_stream["id"], raw_event=event)
            )
            self.messages_in_process[self.active_run["id"]] = None
            return

        if is_tool_call_start_event and should_emit_tool_calls:
            yield self._dispatch_event(
                ToolCallStartEvent(
                    type=EventType.TOOL_CALL_START,
                    tool_call_id=tool_call_data["id"],
                    tool_call_name=tool_call_data["name"],
                    parent_message_id=event["data"]["chunk"].id,
                    raw_event=event,
                )
            )
            self.set_message_in_progress(
                self.active_run["id"],
                MessageInProgress(id=event["data"]["chunk"].id, tool_call_id=tool_call_data["id"],
                                  tool_call_name=tool_call_data["name"])
            )
            return

        if is_tool_call_args_event and should_emit_tool_calls:
            yield self._dispatch_event(
                ToolCallArgsEvent(
                    type=EventType.TOOL_CALL_ARGS,
                    tool_call_id=current_stream["tool_call_id"],
                    delta=tool_call_data["args"],
                    raw_event=event
                )
            )
            return

        if is_message_content_event and should_emit_messages:
            if bool(current_stream and current_stream.get("id")) == False:
                yield self._dispatch_event(
                    TextMessageStartEvent(
                        type=EventType.TEXT_MESSAGE_START,
                        role="assistant",
                        message_id=event["data"]["chunk"].id,
                        raw_event=event,
                    )
                )
                self.set_message_in_progress(
                    self.active_run["id"],
                    MessageInProgress(
                        id=event["data"]["chunk"].id,
                        tool_call_id=None,
                        tool_call_name=None
                    )
                )
                current_stream = self.get_message_in_progress(self.active_run["id"])

            yield self._dispatch_event(
                TextMessageContentEvent(
                    type=EventType.TEXT_MESSAGE_CONTENT,
                    message_id=current_stream["id"],
                    delta=message_content,
                    raw_event=event,
                )
            )
            return

    elif event_type == LangGraphEventTypes.OnChatModelEnd:
        if self.get_message_in_progress(self.active_run["id"]) and self.get_message_in_progress(
                self.active_run["id"]).get("tool_call_id"):
            resolved = self._dispatch_event(
                ToolCallEndEvent(type=EventType.TOOL_CALL_END,
                                 tool_call_id=self.get_message_in_progress(self.active_run["id"])["tool_call_id"],
                                 raw_event=event)
            )
            if resolved:
                self.messages_in_process[self.active_run["id"]] = None
            yield resolved
        elif self.get_message_in_progress(self.active_run["id"]) and self.get_message_in_progress(
                self.active_run["id"]).get("id"):
            resolved = self._dispatch_event(
                TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END,
                                    message_id=self.get_message_in_progress(self.active_run["id"])["id"],
                                    raw_event=event)
            )
            if resolved:
                self.messages_in_process[self.active_run["id"]] = None
            yield resolved

    elif event_type == LangGraphEventTypes.OnCustomEvent:
        if event["name"] == CustomEventNames.ManuallyEmitMessage:
            yield self._dispatch_event(
                TextMessageStartEvent(type=EventType.TEXT_MESSAGE_START, role="assistant",
                                      message_id=event["data"]["message_id"], raw_event=event)
            )
            yield self._dispatch_event(
                TextMessageContentEvent(
                    type=EventType.TEXT_MESSAGE_CONTENT,
                    message_id=event["data"]["message_id"],
                    delta=event["data"]["message"],
                    raw_event=event,
                )
            )
            yield self._dispatch_event(
                TextMessageEndEvent(type=EventType.TEXT_MESSAGE_END, message_id=event["data"]["message_id"],
                                    raw_event=event)
            )

        elif event["name"] == CustomEventNames.ManuallyEmitToolCall:
            yield self._dispatch_event(
                ToolCallStartEvent(
                    type=EventType.TOOL_CALL_START,
                    tool_call_id=event["data"]["id"],
                    tool_call_name=event["data"]["name"],
                    parent_message_id=event["data"]["id"],
                    raw_event=event,
                )
            )
            yield self._dispatch_event(
                ToolCallArgsEvent(
                    type=EventType.TOOL_CALL_ARGS,
                    tool_call_id=event["data"]["id"],
                    delta=event["data"]["args"] if isinstance(event["data"]["args"], str) else json.dumps(
                        event["data"]["args"]),
                    raw_event=event
                )
            )
            yield self._dispatch_event(
                ToolCallEndEvent(type=EventType.TOOL_CALL_END, tool_call_id=event["data"]["id"], raw_event=event)
            )

        elif event["name"] == CustomEventNames.ManuallyEmitState:
            self.active_run["manually_emitted_state"] = event["data"]
            yield self._dispatch_event(
                StateSnapshotEvent(type=EventType.STATE_SNAPSHOT,
                                   snapshot=self.get_state_snapshot(self.active_run["manually_emitted_state"]),
                                   raw_event=event)
            )

        yield self._dispatch_event(
            CustomEvent(type=EventType.CUSTOM, name=event["name"], value=event["data"], raw_event=event)
        )

    elif event_type == LangGraphEventTypes.OnToolEnd:
        tool_call_output = event["data"]["output"]

        # ============ PATCH START ============
        
        if isinstance(tool_call_output, Command):
            # Extract ToolMessages from Command.update
            messages = tool_call_output.update.get('messages', [])
            tool_messages = [m for m in messages if isinstance(m, ToolMessage)]

            # Process each tool message
            for tool_msg in tool_messages:
                if not self.active_run["has_function_streaming"]:
                    yield self._dispatch_event(
                        ToolCallStartEvent(
                            type=EventType.TOOL_CALL_START,
                            tool_call_id=tool_msg.tool_call_id,
                            tool_call_name=tool_msg.name,
                            parent_message_id=tool_msg.id,
                            raw_event=event,
                        )
                    )
                    yield self._dispatch_event(
                        ToolCallArgsEvent(
                            type=EventType.TOOL_CALL_ARGS,
                            tool_call_id=tool_msg.tool_call_id,
                            delta=json.dumps(event["data"].get("input", {})),
                            raw_event=event
                        )
                    )
                    yield self._dispatch_event(
                        ToolCallEndEvent(
                            type=EventType.TOOL_CALL_END,
                            tool_call_id=tool_msg.tool_call_id,
                            raw_event=event
                        )
                    )

                yield self._dispatch_event(
                    ToolCallResultEvent(
                        type=EventType.TOOL_CALL_RESULT,
                        tool_call_id=tool_msg.tool_call_id,
                        message_id=str(uuid.uuid4()),
                        content=tool_msg.content,
                        role="tool"
                    )
                )
            return
        # ============ PATCH END ============

        # Original behavior for non-Command objects
        if not self.active_run["has_function_streaming"]:
            yield self._dispatch_event(
                ToolCallStartEvent(
                    type=EventType.TOOL_CALL_START,
                    tool_call_id=tool_call_output.tool_call_id,
                    tool_call_name=tool_call_output.name,
                    parent_message_id=tool_call_output.id,
                    raw_event=event,
                )
            )
            yield self._dispatch_event(
                ToolCallArgsEvent(
                    type=EventType.TOOL_CALL_ARGS,
                    tool_call_id=tool_call_output.tool_call_id,
                    delta=json.dumps(event["data"]["input"]),
                    raw_event=event
                )
            )
            yield self._dispatch_event(
                ToolCallEndEvent(
                    type=EventType.TOOL_CALL_END,
                    tool_call_id=tool_call_output.tool_call_id,
                    raw_event=event
                )
            )

        yield self._dispatch_event(
            ToolCallResultEvent(
                type=EventType.TOOL_CALL_RESULT,
                tool_call_id=tool_call_output.tool_call_id,
                message_id=str(uuid.uuid4()),
                content=tool_call_output.content,
                role="tool"
            )
        )


def apply_patch():
    try:
        from ag_ui_langgraph.agent import LangGraphAgent
        if not hasattr(LangGraphAgent, '_handle_single_event_original'):
            LangGraphAgent._handle_single_event_original = LangGraphAgent._handle_single_event
        LangGraphAgent._handle_single_event = patched_handle_single_event

        print("✅ Successfully patched ag_ui_langgraph for LangGraph Command support")
        return True

    except ImportError as e:
        print(f"❌ Could not import ag_ui_langgraph: {e}")
        return False
    except AttributeError as e:
        print(f"❌ Could not find method to patch: {e}")
        return False
    except Exception as e:
        print(f"❌ Unexpected error applying patch: {e}")
        import traceback
        traceback.print_exc()
        return False

apply_patch()

this is a very simple and dirty fix so looking to improve it still

[xdave]

(UPDATE: I'm an idiot and and fixed my repro below)

I'm having this exact problem and the above patch does work. Here's how to reproduce:

pyproject.toml

[project]
name = "issue-431-repro"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "ag-ui-langgraph>=0.0.18",
    "fastapi>=0.120.3",
    "langchain>=1.0.3",
    "langchain-core>=1.0.2",
    "langchain-openai>=1.0.1",
    "langgraph>=1.0.2",
    "uvicorn>=0.38.0",
]

main.py

import logging
import os
from typing import Annotated

import uvicorn
from ag_ui_langgraph import LangGraphAgent, add_langgraph_fastapi_endpoint
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from langchain.agents import AgentState, create_agent
from langchain.messages import ToolMessage
from langchain.tools import InjectedToolCallId, tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import InjectedState
from langgraph.types import Command, RunnableConfig
from pydantic import BaseModel

logging.basicConfig(level=logging.INFO)

class SearchResults(BaseModel):
    """Search results"""

    results: list[str]


@tool
def web_search(query: str) -> SearchResults:
    """Perform a web search to find information."""

    return SearchResults(results=[
        "Result 1: Example search result",
        "Result 2: Another example search result"
    ])

model = ChatOpenAI(
    model="gpt-4o",
    api_key=os.getenv("OPENAI_API_KEY")
)

search_agent = create_agent(
    name="search_agent",
    model=model,
    tools=[web_search],
    system_prompt="""You are a Web Search specialist.""",
    checkpointer=MemorySaver()
)

def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Transfer to {agent_name}"

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[AgentState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = ToolMessage(
            role="tool",
            content=f"Successfully transferred to {agent_name}",
            name=name,
            tool_call_id=tool_call_id,
        )
        return Command(
            goto=agent_name,
            update={**state, "messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )

    return handoff_tool

transfer_to_search_agent = create_handoff_tool(
    agent_name="search_agent",
    description="Transfer to Web Search Agent to find information on the web.",
)

supervisor = create_agent(
    name="supervisor",
    model=model,
    tools=[transfer_to_search_agent],
    system_prompt="""You are the Supervisor Agent.""",
    checkpointer=MemorySaver()
)

supervisor_graph = (
    StateGraph(AgentState)
    .add_node(supervisor, destinations=("search_agent", END))
    .add_node(search_agent)
    .add_edge(START, "supervisor")
    .add_edge("search_agent", END)
    .compile(checkpointer=MemorySaver())
)

agent = LangGraphAgent(
    name="supervisor",
    description="Supervisor agent that routes to specialized sub-agents.",
    graph=supervisor_graph,
    config=RunnableConfig(recursion_limit=1000),
)

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

add_langgraph_fastapi_endpoint(app, agent, "/agent")

def main():
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run('main:app', host="0.0.0.0", port=port, reload=True)

if __name__ == "__main__":
    main()

How to run

• uv run main.py
• connect to http://localhost:8000 with ag-ui client
• ask to search for something

Result (without patch)

• Same as OP

Result (with above patch)

• works

After downgrading LangChain/LangGraph to pre-1.0.0

• Same