RetrievalGraph result cannot be streamed in open-agent-platform frontend

[xlcaptain]

**Describe **

When I use the open-agent-platform frontend to call my custom LangGraph (RetrievalGraph), the response from the backend is in the format:

event: messages|respond:5ecec403-30a5-ebd1-8c53-4745946ec2df
data:  xxxxxx

This prevents the frontend from streaming the output to the user in real time. Instead, the result is only available after the entire process is finished, which is not the expected streaming behavior.

---

To Reproduce

1. I have implemented a custom LangGraph as follows (simplified for clarity):

from langgraph.graph import StateGraph, START, END
from laws.state import AgentState, InputState
from laws.researcher_graph.graph import graph as researcher_graph
from laws.utils import format_docs, load_chat_model
from langchain_core.runnables import RunnableConfig
from pydantic import BaseModel, Field
from laws import prompt as _prompt
from api.config import model_32b_nothink_dev, model_32b, openai_model
from laws.parser import StructuredExtractor
from typing import Any, Literal
from langchain_core.messages.base import BaseMessage

# 1. Generate research plan
async def create_research_plan(state: AgentState, config: RunnableConfig) -> dict[str, list[str]]:
    class Plan(BaseModel):
        steps: list[str] = Field(default_factory=list, description="Generate a research plan based on the user's question")
    extractor = StructuredExtractor(schema=Plan, llm=model_32b_nothink_dev)
    messages = [{"role": "system", "content": _prompt.RESEARCH_PLAN_SYSTEM_PROMPT}] + state.messages
    response = await extractor.aparse(messages)
    steps = response.steps + [state.messages[-1].content]
    return {
        "steps": steps,
        "documents": "delete",
        "query": state.messages[-1].content,
    }

# 2. Stepwise retrieval
async def conduct_research(state: AgentState, config: RunnableConfig) -> dict[str, Any]:
    if not state.steps:
        return {}
    result = await researcher_graph.ainvoke({"question": state.steps[0]})
    return {"documents": result["documents"], "steps": state.steps[1:]}

# 3. Check if finished
def check_finished(state: AgentState) -> Literal["respond", "conduct_research"]:
    if state.steps:
        return "conduct_research"
    else:
        return "respond"

# 4. Generate final answer
async def respond(state: AgentState, config: RunnableConfig) -> dict[str, list[BaseMessage]]:
    top_k = 20
    context = format_docs(state.documents[:top_k])
    prompt = _prompt.RESPONSE_SYSTEM_PROMPT.format(context=context)
    messages = [{"role": "system", "content": prompt}] + state.messages
    response = await openai_model.ainvoke(messages)
    return {"messages": [response], "answer": response.content}

# Build StateGraph
builder = StateGraph(AgentState, input=InputState)
builder.add_node(create_research_plan)
builder.add_node(conduct_research)
builder.add_node(respond)
builder.add_edge(START, "create_research_plan")
builder.add_edge("create_research_plan", "conduct_research")
builder.add_conditional_edges("conduct_research", check_finished)
builder.add_edge("respond", END)
graph = builder.compile()
graph.name = "RetrievalGraph"

2. I deploy this graph using open-agent-platform and call it from the frontend using the following API:

POST http://192.xx.xx.xx:8124/threads/<thread_id>/runs/stream
{
    "input": {
        "messages": [
            {
                "id": "000d8c0d-b185-4a94-b76e-52a37b42c7a9",
                "type": "human",
                "content": [
                    {
                        "type": "text",
                        "text": "hello world"
                    }
                ]
            }
        ]
    },
    "config": {
        "configurable": {
            "apiKeys": {
                "OPENAI_API_KEY": "",
                "ANTHROPIC_API_KEY": "",
                "GOOGLE_API_KEY": "",
                "TAVILY_API_KEY": ""
            }
        }
    },
    "metadata": {},
    "stream_mode": [
        "values",
        "messages-tuple",
        "custom"
    ],
    "stream_subgraphs": true,
    "stream_resumable": true,
    "assistant_id": "<assistant_id>",
    "on_disconnect": "continue"
}

3. The response I get is:

event: messages|respond:5ecec403-30a5-ebd1-8c53-4745946ec2df

Expected behavior

I expect the frontend to be able to stream the generated content in real time, with SSE events of the form:

event: messages
data: ...

just like in the official examples, rather than receiving events like:

event: messages|respond:5ecec403-30a5-ebd1-8c53-4745946ec2df
data: [{"content":"hello","additional_kwargs":{} ...
id: 115

event: messages|respond:5ecec403-30a5-ebd1-8c53-4745946ec2df
data: [{"content":"world","additional_kwargs":{} ...
id: 116

which prevents the frontend from displaying the output in a true streaming fashion.

---

Additional context

• The respond node in my graph is asynchronous and returns a list of BaseMessage.

---

Questions

• What should I change in my graph structure or return values to ensure the SSE event type is messages (not messages|respond:...) so that the frontend can stream the output as expected?
• Is there a specific return type, node naming convention, or protocol required for open-agent-platform to emit event: messages?
• How can I modify my graph or API usage to achieve real-time streaming output with the correct event type?
• Could the issue be related to the stream_mode configuration? For example, does setting stream_mode to "values", "messages-tuple", or "custom" affect the SSE event type or the way messages are streamed to the frontend? If so, what is the recommended stream_mode setting to ensure the event type is always messages?

---

Thank you for your help!

[King-Techno]

I’m facing the same issue. Keen to get a solution. @xlcaptain did you manage to resolve it?