Custom Graphs¶

Each GraphRegistry key becomes an OpenAI model name.

Default Message Graph¶

GraphConfig(
    graph=my_graph,
    description="Answer questions with the default message graph.",
    streamable_node_names=["generate"],
)

Without adapters, graph input is {"messages": langchain_messages} and the last output message must be an AIMessage. The required description is published in LGOS model list and detail extensions for catalog UIs.

Message Ownership¶

Return an AIMessage only from the node or subgraph that owns the final assistant turn. Internal workers should return structured state; status and progress should use custom events. add_messages preserves message history but does not enable streaming or combine multiple assistant messages.

Custom Schemas¶

Use adapters when your graph has native LangGraph input, output, or context schemas:

GraphConfig(
    graph=custom_io_graph,  # (1)!
    description="Answer questions with application context.",
    request_to_input=request_to_input,  # (2)!
    context_factory=context_factory,  # (3)!
    output_to_message=output_to_message,  # (4)!
)

1. Keep the graph's native LangGraph schema.
2. Build graph input from the validated OpenAI request and converted messages.
3. Build optional LangGraph runtime context from the request and public settings.
4. Render the graph's native output as a durable LangChain AIMessage.

See demo/api/src/lgos_demo_api/graphs/custom_io.py for the runnable version.

Runtime Context¶

Use LangGraph runtime context for immutable, per-invocation application values that nodes need but that do not belong in graph state. Define a context schema, declare it on StateGraph, and read it from the injected Runtime object:

from dataclasses import dataclass

from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from typing_extensions import TypedDict


@dataclass(frozen=True)
class AppContext:
    user_id: str


class State(TypedDict, total=False):
    question: str
    answer: str


async def generate(state: State, runtime: Runtime[AppContext]) -> dict[str, str]:
    return {
        "answer": f"{runtime.context.user_id} asked: {state['question']}"
    }


custom_graph = (
    StateGraph(State, context_schema=AppContext)
    .add_node("generate", generate)
    .add_edge(START, "generate")
    .add_edge("generate", END)
    .compile()
)

Build that context from the validated OpenAI request at the adapter boundary:

from langchain_core.messages import AIMessage, BaseMessage

from langgraph_openai_serve import ClientSettings, GraphConfig
from langgraph_openai_serve.api.chat.schemas import ChatCompletionRequest


def request_to_input(
    request: ChatCompletionRequest,
    messages: list[BaseMessage],
) -> State:
    return {"question": str(messages[-1].content or "")}


def context_factory(
    request: ChatCompletionRequest,
    _client_settings: ClientSettings | None,
) -> AppContext:
    return AppContext(user_id=request.user or "anonymous")


def output_to_message(output: State) -> AIMessage:
    return AIMessage(content=output["answer"])


custom_graph_config = GraphConfig(
    graph=custom_graph,
    description="Answer questions with application context.",
    request_to_input=request_to_input,
    context_factory=context_factory,
    output_to_message=output_to_message,
)

LGOS passes the returned value through LangGraph's context argument. Do not put application values such as user_id, model selection, or prompt options in config["configurable"].

Runtime Settings¶

Expose only a safe, explicit subset of runtime context when an ordinary OpenAI client should configure a graph. Define that public subset as a ClientSettings model. All fields automatically share one JSON metadata envelope:

from pydantic import Field

from langgraph_openai_serve import ClientSettings


class PublicRuntimeSettings(ClientSettings):
    use_history: bool = Field(
        default=True,
        title="Use conversation history",
    )

When public settings are the complete runtime context, declare the same class as the graph's context_schema. In every case, set client_settings=PublicRuntimeSettings on the compiled graph's GraphConfig. LGOS validates the request directly from JSON and, without a context_factory, passes the resulting PublicRuntimeSettings instance as LangGraph runtime context. Graph authors must keep the inherited strict, frozen, extra-forbid, and default-validation behavior; LGOS rejects a settings model that changes it.

Follow Configure LangGraph Runtime Settings for discovery, request transport, and per-request behavior. The runnable demo version is in demo/api/src/lgos_demo_api/graphs/simple.py.

Async Factories¶

GraphConfig.graph may be a compiled graph, sync factory, or async factory:

async def advanced_graph():
    tools = await mcp_client.get_tools()
    return create_agent(model=model, tools=tools)

GraphConfig(
    graph=advanced_graph,
    description="Answer questions with asynchronously loaded tools.",
)

See demo/api/src/lgos_demo_api/graphs/advanced_mcp.py for a mock MCP-style example.

Register And Bind¶

from langgraph_openai_serve import GraphConfig, GraphRegistry, LanggraphOpenaiServe

graphs = GraphRegistry(
    registry={
        "my-graph": GraphConfig(
            graph=my_graph,
            description="Answer questions with my graph.",
            streamable_node_names=["generate"],
        ),
        "advanced-mcp-tools": GraphConfig(
            graph=advanced_graph,
            description="Answer questions with MCP tools.",
        ),
    }
)

LanggraphOpenaiServe(graphs=graphs).bind_openai_api()

Streaming¶

When an OpenAI request sets stream=True, LGOS forwards only streamed AIMessageChunk values from streamable_node_names. A graph with no eligible chunks still receives its final rendered AIMessage after execution, so a caller does not need a separate non-streaming code path.

When several public nodes contribute text, return their completed messages through the graph's messages channel and make output_to_message render the same ordered content for a complete response.

Status Updates¶

Publish meaningful status from long-running graph code:

from langgraph.config import get_stream_writer
from langgraph_openai_serve import GraphConfig, GraphFeature, status_event


async def generate_audio(state):
    writer = get_stream_writer()
    writer(status_event("Generating audio"))

    audio = await audio_service.generate(state["text"])

    writer(status_event("Audio ready", done=True))
    return {"audio": audio}


status_graph_config = GraphConfig(
    graph=status_graph,
    description="Generate audio with visible status updates.",
    features={GraphFeature.CLIENT_EVENTS},
)

Declare GraphFeature.CLIENT_EVENTS on every graph that emits these events. Streaming clients then opt into the events with metadata={"langgraph_stream_events": "v1"}. Emit a final done=True update so native clients stop showing the status as active. Use hidden=True on that final update when the status should disappear after completion.

These passive updates are not OpenAI tool calls, which would ask the client to execute work. The graph remains responsible for its own work; the client only renders status.

Interrupts¶

Enable the interrupt feature for checkpointed human-in-the-loop graphs:

from langgraph_openai_serve import GraphConfig, GraphFeature
from langgraph_openai_serve.graph.interrupt import InMemoryRunCoordinator

GraphConfig(
    graph=interruptible_graph,
    description="Collect human input before performing an action.",
    features={GraphFeature.INTERRUPTS},
    run_coordinator=InMemoryRunCoordinator(),
)

The graph must be compiled with an asynchronous checkpointer that implements aget_tuple(), alist(), aput(), aput_writes(), and adelete_thread(). LGOS generates a UUID for an initial interrupt run; callers only need to send metadata.langgraph_run_id when they want to choose that UUID for deterministic retries and isolation. The OpenAI tool-call ID and opaque arguments carry the operation and state-generation identities needed for a resume.

Clients must preserve the complete assistant tool_calls message and submit exactly one result for every pending call in one resume request. See Interrupt resume for client code and Tool calls and interrupts for the normative protocol, node-restart/idempotency rules, and retention requirements. Harden persistent deserialization according to LangGraph's security advisory.

Next Steps¶

• OpenAI clients
• LangGraph integration
• Reference