LangGraph Integration¶

A registered graph name becomes the OpenAI model value clients pass to /v1/responses or /v1/chat/completions.

Registration¶

GraphRegistry stores model names and GraphConfig values:

GraphRegistry(
    registry={
        "chat": GraphConfig(
            graph=chat_graph,
            description="General-purpose chat graph.",
        ),
        "mcp-mock": GraphConfig(
            graph=mcp_mock_graph,
            description="Chat graph with asynchronously loaded tools.",
        ),
    }
)

GraphConfig.graph can be a compiled graph, sync factory, or async factory. Async factories support setup such as MCP-style tool discovery before creating a compiled graph.

Adaptation¶

The default graph contract is:

• input: {"messages": langchain_messages}
• output: the last AIMessage in result["messages"]

Use request_to_input, context_factory, and output_to_message when the graph has custom input, output, or context schemas. Those adapters keep the public HTTP surface OpenAI-compatible while letting the graph stay idiomatic LangGraph.

Graphs that use LangGraph's MessagesState (or a messages channel reduced by add_messages) explicitly decide which completed messages belong to their conversation state. Shared-key subgraphs can update that channel directly; private subgraphs must map selected results back in their wrapper. This state choice does not, by itself, select live token streaming or concatenate several messages into one OpenAI response.

Context Versus Config¶

LGOS keeps LangGraph's invocation channels separate:

graph.ainvoke(  # use graph.astream(...) for streaming
    graph_input,
    context=runtime_context,
    config=runnable_config,
)

• graph_input contains mutable workflow state, including converted messages.
• runtime_context starts with optional validated GraphConfig.client_settings settings and can be composed with server-owned values by GraphConfig.context_factory(request, settings). Settings used directly must also be the graph's context_schema; every non-null factory result requires a context schema. LangGraph applies its native schema coercion when it runs the graph. Nodes receive the result as runtime.context on an injected Runtime[Context].
• runnable_config contains callbacks and execution identity. For an interrupt-enabled graph, LGOS derives config["configurable"]["thread_id"] from the server's checkpoint scope, registered model, and operation UUID. The UUID is generated by the server unless the initial request supplies metadata.lgos_run_id. When callbacks are configured, LGOS also adds a stable run name and request/model/operation correlation metadata for tracing. LangGraph also propagates the primitive configurable checkpoint thread_id to callbacks during interrupt-enabled execution.

Application settings that nodes consume belong in typed runtime context, not in the configurable section of RunnableConfig. The internal checkpoint key is different: the checkpointer needs it to restore state before node execution, so it remains execution configuration. It is intentionally scoped by a server-trusted value plus model and run; it is not derived from a UI conversation ID.

Because LGOS supports Python 3.11 and newer, callback/config context propagates automatically to nested async runnable calls. Node functions only need an injected RunnableConfig when they inspect or modify execution configuration; they do not need one solely to pass config to a nested model's ainvoke().

See Custom Graphs for a typed server-owned context example and Configure LangGraph Runtime Settings for public settings, discovery, and request handling. LangGraph's official runtime, streaming, and persistence documentation for the underlying conventions.

Runner Behavior¶

LGOS uses LangGraph's stable v2 output wrapper so interrupt handling remains durable. Ordinary requests choose invocation or streaming according to the OpenAI response mode; server tools additionally need intermediate updates.

Internal ChatOpenAI calls that must not reach the assistant text stream set disable_streaming=True. Graph authors must follow the assistant text parity contract because a graph cannot retract an intermediate draft after it has streamed it.

Interrupt runs use exit durability and drain graph execution before exposing a durable tool-call batch. A GraphConfig.run_coordinator lease covers state inspection, validation, and execution; busy leases fail with HTTP 409. LGOS retains only checkpoints it has exposed as an interrupt and otherwise performs best-effort terminal cleanup. The complete lifecycle is specified in OpenAI compatibility.

One prepared-run async context owns that lease and its checkpoint disposition. State is left untouched until execution starts, becomes cleanup-eligible while execution is incomplete, and becomes retained only when the runner commits a validated interrupt batch. The context deletes terminal or incomplete temporary state before releasing the lease. Cleanup is idempotent, shielded from outer request cancellation, and never replaces an execution or cancellation failure.

The graph therefore needs an async checkpointer implementing aget_tuple(), alist(), aput(), aput_writes(), and adelete_thread(), plus a coordinator shared by all workers. See PostgreSQL Coordination for the production adapter.

Paused Runs Across Deployments¶

Paused state resumes against the graph implementation currently selected by the model. Drain, expire, migrate, or route pending operations before changing state schemas, pending nodes, or interrupt order. Keep the checkpoint scope stable when a release must resume old state; otherwise keep the old release routable. See LangGraph's official backward-compatibility contract.

Status updates use LangGraph's native custom stream mode. No middleware is needed. Graph code emits a status exactly where it has enough application context to describe the long-running work in user-facing language.

See Custom Graphs for runnable examples.

Direct Runner Calls¶

Python callers use run_langgraph(request, messages, registry) or run_langgraph_stream(request, messages, registry) from langgraph_openai_serve.graph.runner. Both accept a protocol-neutral GraphRequest and a list of LangChain BaseMessage values. HTTP request decoding belongs to the corresponding API adapter; the graph runner imports neither API's request types.

run_langgraph() returns the final AIMessage or LangGraphInterruptBatch directly. The streaming helper yields text and custom events followed by that same output type. These public helpers own the prepared-run context for their complete lifetime. Lower-level invoke_run() and stream_run() calls operate only inside the active context supplied by an HTTP service or direct wrapper.

When continuing a paused run, pass the decoded InterruptResume as resume=. Pass a server-trusted checkpoint_scope= consistently on the initial invocation and every resume. The helpers delegate to the same prepare_run(), invoke_run(), and stream_run() used by both HTTP routes.

The demo's api/notebooks/graph_runner.py compares direct execution with Responses SDK calls. Open it with just demo/marimo --editable.

Request Cancellation¶

For streaming Responses and Chat Completions (stream=true), LGOS ties graph iteration to the HTTP response lifetime. A request-scoped FastAPI dependency owns the producer task and memory channel behind StreamingResponse. When the client disconnects, dependency cleanup cancels and awaits that producer, then closes the graph iterator. The service's prepared-run context owns graph cleanup after streaming starts; the request owner releases the run itself when a stream is closed before its source starts. This uses the normal OpenAI streaming connection; LGOS adds no custom cancellation route, header, or SSE event.