Reference¶

OpenAI-Compatible API¶

Default prefix: /v1. Change it with LGOS_OPENAI_API_PREFIX or bind_openai_api(prefix=...). Generic access logs are emitted by the deployment's ASGI server or ingress proxy.

FastAPI docs for the mounted OpenAI app are disabled by default. Set LGOS_OPENAI_API_DOCS_ENABLED=true to expose {prefix}/docs, {prefix}/redoc, and {prefix}/openapi.json.

Settings¶

Package settings:

Settings prefixed with DEMO_ belong to the independent example applications and are documented under Demo Settings and Commands.

Public API¶

Use LanggraphOpenaiServe to bind OpenAI-compatible routes to a FastAPI app. After binding, server.openai_app exposes the mounted FastAPI application for host integrations such as manual middleware or telemetry instrumentation. Use GraphRegistry to map OpenAI model names to GraphConfig values. The registry must contain at least one graph. Pydantic rejects empty registries and model IDs that cannot be addressed as one URL path segment. Registry keys are read-only after validation; use registry.register(model_id, config) to add or replace a graph.

LanggraphOpenaiServe(..., checkpoint_scope=resolver) accepts an optional sync or async callable from FastAPI Request to a non-empty, server-trusted string. Interrupt checkpoint keys include this scope before model and run identity. Use an authenticated tenant or principal identifier when caller-chosen run UUIDs must be isolated between security domains; do not derive the scope from untrusted Chat Completions metadata or the Chat Completions user field. The default "default" scope is suitable only for a single-tenant or shared-trust deployment. The resolver must return the same scope for the initial request and its resume; changing tenant identity makes the other scope's checkpoint deliberately unreachable.

GraphConfig accepts:

• graph: compiled graph, sync factory, or async factory.
• description: required human-readable model description advertised by model listing and retrieval.
• streamable_node_names: node names whose streamed AIMessageChunk values are forwarded as assistant text. If several nodes contribute, the graph's output adapter must render the same ordered content for complete responses.
• features: GraphFeature values that enable optional server behavior.
• client_settings: explicit public ClientSettings model class advertised by model retrieval.
• runtime_callbacks: callbacks included in the LangGraph RunnableConfig. When Langfuse tracing is enabled, LGOS adds its callback without mutating this collection or manager.
• run_coordinator: asynchronous single-flight coordination for interrupt runs. It rejects an occupied LGOS checkpoint key instead of queueing it and returns an async context manager.
• request_to_input(request, messages): custom OpenAI request to graph input.
• context_factory(request, client_settings): compose the final typed LangGraph runtime context from server-owned values and optional validated public settings.
• output_to_message(output): custom graph output to a durable AIMessage.

When both are configured, LGOS validates the public settings first and passes them to context_factory. Without a factory, the validated settings instance is the runtime context, so the graph must use that settings model as its context_schema. A factory may return None; every non-null result requires a graph context schema. LGOS passes server-owned factory results to LangGraph without rebuilding them. LangGraph's native runtime-context handling constructs mapping values through dataclass and Pydantic context schemas and trusts existing instances. The factory owns the validity of instances it creates. Graphs should access context from an injected Runtime[Context]. Runtime context is separate from RunnableConfig:

LGOS assembles runnable config from runtime_callbacks and, for an interrupt-enabled run, a fixed-length SHA-256 checkpoint key derived from the server-trusted scope, registered model, and operation UUID. This is deliberately not a UI chat or thread ID. There is intentionally no adapter for placing arbitrary OpenAI request fields into config["configurable"]; use typed runtime context for values consumed by nodes.

Langfuse is a first-class optional integration. Install it and enable the default callback through process environment settings:

uv add "langgraph-openai-serve[tracing]"
export LGOS_ENABLE_LANGFUSE=true
export LANGFUSE_PUBLIC_KEY=pk-lf-...
export LANGFUSE_SECRET_KEY=sk-lf-...

LANGFUSE_BASE_URL is optional; Langfuse Cloud is the default. Set it only for a different cloud region or a self-hosted instance. Langfuse's CallbackHandler owns its standard SDK configuration and error behavior. LGOS constructs it on the first graph run that needs runnable configuration, then reuses that process-wide handler. When enabled, the deployment-level toggle is authoritative: LGOS adds Langfuse alongside empty, list, or manager callbacks without altering the registered GraphConfig or caller-owned collection. To provide a custom Langfuse handler, leave the toggle off and pass that handler through runtime_callbacks.

For explicit construction, import langgraph_openai_serve.integrations.langfuse.get_langfuse_callback or pass an application-created vendor handler through runtime_callbacks.

When a callback is present, LGOS gives the graph run the stable name lgos.chat_completion and adds RunnableConfig.metadata fields for the request ID, registered graph model, (for interrupt runs) operation ID, and (when the request supplies metadata.session_id) the Langfuse-recognized langfuse_session_id. LangGraph also propagates primitive configurable values during execution, so callbacks on interrupt runs receive the derived checkpoint thread_id. LGOS does not set LangChain's native tracer run_id or force a custom Langfuse trace ID. See Production Logging and Request Correlation.

The same features set drives runtime behavior and the versioned langgraph_openai_serve.features extension returned by GET /v1/models/{model}. GraphFeature.CLIENT_EVENTS enables and advertises public client-event chunks. GraphFeature.INTERRUPTS enables and advertises the interrupt/resume flow.

Runtime Settings¶

Subclass ClientSettings to publish only fields deliberately selected by the server author. LGOS never inspects or publishes the LangGraph context schema:

from pydantic import Field

from langgraph_openai_serve import ClientSettings


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

Pass this model as GraphConfig.client_settings and use it as the graph's context schema when it is the complete runtime context. Every public field must have a default. Registration rejects subclasses that change the inherited strict, frozen, extra-forbid, or default-validation behavior, as well as fields excluded from Pydantic serialization.

All public fields travel together as compact JSON text in the metadata.langgraph_runtime_settings string. Clients omit values equal to the advertised defaults. System instructions remain ordinary OpenAI messages and are independent of ClientSettings; native Chat Completions fields keep their standard request semantics.

LGOS validates defaults and generates the discovery JSON Schema when the graph is registered, then validates settings on every request. Without context_factory, the settings become Runtime.context. A factory can instead combine them with server-derived identity, authorization, database clients, and other dependencies.

The serialized descriptor appears only on model retrieval as langgraph_openai_serve.client_settings, with independent schema_version, json_schema, and defaults fields. All client settings use the fixed metadata.langgraph_runtime_settings envelope. Clients use the descriptor's validated defaults object as the baseline; default keywords within the generated JSON Schema are annotations, not the runtime baseline.

See Configure LangGraph Runtime Settings for the runtime settings flow, and Runtime Settings for the request lifecycle.

Interrupt-enabled graphs have additional registration requirements:

• compile the graph with an asynchronous checkpointer that supports aget_tuple(), alist(), aput(), aput_writes(), and adelete_thread();
• configure an asynchronous run_coordinator; and
• use a durable checkpointer and cross-process coordinator in production.

The initial request does not require metadata. LGOS generates a UUID operation ID and returns it in every interrupt tool call. A caller that needs deterministic initial-request retries can instead supply a non-nil UUID in metadata.langgraph_run_id. InMemoryRunCoordinator is suitable only for tests and a single-process development server; it cannot serialize requests across workers or hosts.

Pending checkpoints exist only to resume an interrupt batch returned to the client. LGOS deletes isolated checkpoint state after terminal completion or when execution fails or is cancelled before producing that batch. Operators must separately define an expiry policy for runs abandoned after a batch is returned.

PostgreSQL Coordination¶

Install langgraph-openai-serve[postgres] to use the public langgraph_openai_serve.integrations.postgres.PostgresRunCoordinator. Use LangGraph's official AsyncPostgresSaver for checkpoints and AsyncPostgresStore for application data. The LGOS adapter supplies only the cross-worker interrupt-run lease; it does not replace either storage primitive. Run each configured storage adapter's setup() once before API workers start. A shared pool must follow the upstream connection requirements: autocommit=True, prepare_threshold=0, and mapping rows.

PostgresRunCoordinator(pool, max_concurrent_leases=...) accepts an existing psycopg_pool.AsyncConnectionPool configured with mapping rows and the default close_returns=False; physical session closure is the safety fallback for an indeterminate lock operation. When persistence adapters share that pool, set the lease limit below the pool maximum so at least one connection remains available for persistence I/O. Create one coordinator per process-owned pool so that this capacity limit is not accidentally multiplied. Session advisory locks require direct PostgreSQL connections or session-mode pooling; transaction-mode poolers cannot preserve the lease. Lock contention itself fails immediately through PostgreSQL's pg_try_advisory_lock; connection checkout still follows the pool's configured timeout. The demo deployment uses one pool for both storage adapters and interrupt coordination, plus a separate one-shot schema setup process. Busy interrupt leases fail before streaming begins with HTTP 409 and code: "run_busy".

Client Stream Events¶

Declare the feature on every graph that publishes client events:

from langgraph_openai_serve import GraphConfig, GraphFeature

config = GraphConfig(
    graph=graph,
    description="Graph that reports media-generation status.",
    features={GraphFeature.CLIENT_EVENTS},
)

Inside a long-running graph node or tool, publish user-facing status with status_event():

from langgraph.config import get_stream_writer
from langgraph_openai_serve import status_event

writer = get_stream_writer()
writer(status_event("Generating audio", namespace=("media",)))

# Perform the long-running work.

writer(
    status_event(
        "Audio ready",
        done=True,
        namespace=("media",),
    )
)

The portable status data matches native UI status concepts:

{
  "type": "status",
  "namespace": ["media"],
  "data": {
    "description": "Generating audio",
    "done": false,
    "hidden": false
  }
}

done=False displays ongoing work; always finish a visible status sequence with done=True. Set hidden=True on the final update when clients should remove the status after completion. Status text is deliberately authored by the graph: LGOS does not infer it from internal node names or state.

For other passive notifications, use client_event():

from langgraph.config import get_stream_writer
from langgraph_openai_serve import client_event

get_stream_writer()(
    client_event(
        "progress",
        {
            "stage": "retrieval",
            "completed": 2,
            "total": 5,
            "message": "Searching documents",
        },
        namespace=("research",),
    )
)

The v1 vocabulary is status, progress, and artifact. Event data must be JSON-safe, and every namespace segment must be a string. Keep payloads small and represent large artifacts by an ID or URL. The namespace is a stable, author-defined path; LGOS does not expose LangGraph's dynamic execution namespace.

Events are streaming-only and require both the graph feature and client opt-in. Clients request them with metadata={"langgraph_stream_events": "v1"} and receive a versioned langgraph_openai_serve property on an otherwise standard Chat Completions chunk. Missing and unsupported versions produce the ordinary strict stream. Unknown custom events remain available only to direct runner consumers.

See Client stream events for the wire contract and OpenAI clients for consumption.

Citations¶

Put citations on the final LangChain AIMessage:

from langchain_core.messages import AIMessage
from langchain_core.messages.content import create_citation, create_text_block

message = AIMessage(
    content=[
        create_text_block(
            text="Read the source [1].",
            annotations=[
                create_citation(
                    url="https://example.com/source",
                    title="Example source",
                    start_index=9,
                    end_index=14,
                    cited_text="source",
                )
            ],
        )
    ]
)

Put visible inline citations in the assistant text. Structured annotations add machine-readable provenance; clients are not required to invent marker text from annotation indices.

LangChain citation indices refer to their containing text block. LGOS offsets them into the final response text and preserves OpenAI's inclusive end_index. Use citation_slice(annotation, text) to validate received indices and convert them to a Python slice. LGOS maps native LangChain citations to completed message.annotations; the streaming compatibility extension is added to the final delta.

See Citation ownership for transport and client behavior.

The streaming graph runner preserves LangGraph's native CustomStreamPart values, including their execution namespace. Non-streaming invocation does not subscribe to or replay custom events.

defaults()

30
31
32
33
34
35
36

@classmethod
def defaults(cls) -> Self:
    """Return a deep copy of the registration-validated defaults."""
    return cast(
        "Self",
        _validated_contract(cls).defaults.model_copy(deep=True),
    )

validate_request(request)

38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

@classmethod
def validate_request(cls, request: ChatCompletionRequest) -> Self:
    """Read and validate this model's values from an OpenAI request."""
    parameter = f"metadata.{RUNTIME_SETTINGS_METADATA_KEY}"
    encoded = (request.metadata or {}).get(RUNTIME_SETTINGS_METADATA_KEY, "{}")

    try:
        changes = _validate_json_object(encoded)
    except ValidationError as exc:
        raise ClientSettingsValidationError(
            _validation_message(exc, label="runtime settings"),
            param=parameter,
        ) from exc

    values = client_settings_default_values(cls)
    values.update(changes)

    try:
        settings = cls.model_validate_json(
            _SETTINGS_OBJECT_ADAPTER.dump_json(values),
            strict=True,
            by_alias=False,
            by_name=True,
        )
    except ValidationError as exc:
        field = _first_error_field(exc)
        raise ClientSettingsValidationError(
            _validation_message(exc, label=field or "runtime settings"),
            param=parameter,
        ) from exc
    else:
        _validated_settings_json(settings)
        return settings

build_context(request, graph)

152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

async def build_context(
    self,
    request: ChatCompletionRequest,
    graph: CompiledStateGraph,
) -> Any:
    """Build the LangGraph runtime context for a request."""
    settings = (
        self.client_settings.validate_request(request)
        if self.client_settings is not None
        else None
    )
    if self.context_factory is not None:
        context = await _maybe_await(self.context_factory(request, settings))
    else:
        context = settings

    if context is None:
        return None
    if graph.context_schema is None:
        msg = "A graph that produces runtime context must declare context_schema."
        raise GraphConfigurationError(msg)

    # Preserve server-owned context objects; LangGraph applies context_schema
    # coercion when it invokes the graph.
    return context

build_input(request, messages)

142
143
144
145
146
147
148
149
150

async def build_input(
    self,
    request: ChatCompletionRequest,
    messages: list[BaseMessage],
) -> Any:
    """Build the native graph input for a chat completion request."""
    if self.request_to_input is None:
        return {"messages": messages}
    return await _maybe_await(self.request_to_input(request, messages))

render_output(output)

178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

async def render_output(self, output: Any) -> AIMessage:
    """Convert native graph output into the durable assistant message."""
    if self.output_to_message is not None:
        return await _maybe_await(self.output_to_message(output))

    messages = (
        output["messages"]
        if isinstance(output, Mapping)
        else getattr(output, "messages", None)
    )
    if messages is None:
        msg = "Graph output must expose a messages field."
        raise GraphConfigurationError(msg)
    if not messages:
        return AIMessage(content="")
    message = messages[-1]
    if not isinstance(message, AIMessage):
        msg = "The final graph message must be an AIMessage."
        raise GraphConfigurationError(msg)
    return message

resolve_graph()

101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

async def resolve_graph(self) -> CompiledStateGraph:
    """Get the graph instance, resolving callable graph factories."""
    if self.run_coordinator is not None and not self.supports(
        GraphFeature.INTERRUPTS
    ):
        msg = "run_coordinator is only supported by interrupt-enabled graphs."
        raise GraphConfigurationError(msg)

    if isinstance(self.graph, CompiledStateGraph):
        graph = self.graph
    else:
        graph = await _maybe_await(self.graph())

    if (
        self.client_settings is not None
        and self.context_factory is None
        and graph.context_schema is not self.client_settings
    ):
        msg = (
            "Graphs using client_settings directly must use that settings model "
            "as context_schema."
        )
        raise GraphConfigurationError(msg)

    if self.supports(GraphFeature.INTERRUPTS):
        checkpointer = graph.checkpointer
        if checkpointer is None or any(
            not _overrides_checkpointer_method(checkpointer, method_name)
            for method_name in _INTERRUPT_CHECKPOINTER_METHODS
        ):
            msg = (
                "Interrupt-enabled graphs must use a fully asynchronous "
                "checkpointer with thread deletion."
            )
            raise GraphConfigurationError(msg)
        if self.run_coordinator is None:
            msg = "Interrupt-enabled graphs must configure a run_coordinator."
            raise GraphConfigurationError(msg)

    return graph

supports(feature)

97
98
99

def supports(self, feature: GraphFeature) -> bool:
    """Return whether this graph supports a feature."""
    return feature in self.features

validate_client_settings(value)

88
89
90
91
92
93
94
95

@field_validator("client_settings")
@classmethod
def validate_client_settings(
    cls,
    value: type[ClientSettings] | None,
) -> type[ClientSettings] | None:
    """Validate a public settings model when its graph is registered."""
    return validate_client_settings_model(value) if value is not None else None

get_graph(name)

250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268

def get_graph(self, name: str) -> GraphConfig:
    """
    Get a graph by its name.

    Args:
        name: The name of the graph to retrieve.

    Returns:
        The graph configuration associated with the given name.

    Raises:
        GraphNotFoundError: If the graph name is not found in the registry.

    """
    try:
        return self.registry[name]
    except KeyError as exc:
        msg = f"Graph '{name}' not found in registry."
        raise GraphNotFoundError(msg) from exc

get_graph_names()

246
247
248

def get_graph_names(self) -> list[str]:
    """Get the names of all registered graphs."""
    return list(self.registry.keys())

register(model_id, config)

242
243
244

def register(self, model_id: str, config: GraphConfig) -> None:
    """Add or replace one graph through the validated registry boundary."""
    self.registry = {**self.registry, model_id: config}

LanggraphOpenaiServe(graphs, app=None, checkpoint_scope=None)

 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

def __init__(
    self,
    graphs: GraphRegistry,
    app: FastAPI | None = None,
    checkpoint_scope: Callable[[Request], str | Awaitable[str]] | None = None,
) -> None:
    """
    Initialize the server with a FastAPI app and a populated graph registry.

    Args:
        app: The host FastAPI application to mount the OpenAI API on. If None,
            a new FastAPI app will be created.
        graphs: A GraphRegistry instance containing the graphs to serve.
        checkpoint_scope: Optional server-trusted resolver used to isolate
            interrupt checkpoints by deployment or authenticated principal.

    Raises:
        TypeError: If graphs is not a GraphRegistry instance.

    """
    if not isinstance(graphs, GraphRegistry):
        msg = "Invalid type for graphs parameter. Expected GraphRegistry."
        raise TypeError(msg)

    if app is None:
        app = FastAPI(
            title="LangGraph OpenAI Compatible API",
            description="An OpenAI-compatible API for LangGraph",
            version=get_version(),
        )
    self.app: FastAPI = app
    self._openai_app: FastAPI | None = None
    self.checkpoint_scope = checkpoint_scope or (lambda _request: "default")

    self.graph_registry = graphs

    # Host integrations can inspect registered graphs without traversing the
    # mounted OpenAI sub-application.
    self.app.state.graph_registry = self.graph_registry
    self.app.state.checkpoint_scope = self.checkpoint_scope

    logger.info(
        "server.initialized",
        extra={"graph_count": len(self.graph_registry.registry)},
    )

openai_app

bind_openai_api(prefix=None)

122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

def bind_openai_api(self, prefix: str | None = None) -> "LanggraphOpenaiServe":
    """
    Mount OpenAI-compatible endpoints on the host FastAPI app.

    Args:
        prefix: Optional; The URL prefix for the OpenAI-compatible endpoints.
            Defaults to settings.OPENAI_API_PREFIX.

    """
    prefix = (
        normalize_openai_api_prefix(prefix)
        if prefix is not None
        else settings.OPENAI_API_PREFIX
    )

    openai_app = FastAPI(
        title="LangGraph OpenAI Compatible API",
        description="An OpenAI-compatible API for LangGraph",
        version=get_version(),
        **settings.fastapi_docs_kwargs,
    )
    # Dependencies in mounted routes resolve against the mounted app.
    openai_app.state.graph_registry = self.graph_registry
    openai_app.state.checkpoint_scope = self.checkpoint_scope
    configure_openai_error_handlers(openai_app)
    openai_app.include_router(chat_views.router)
    openai_app.include_router(health_views.router)
    openai_app.include_router(models_views.router)

    self.app.router.routes.append(
        Mount(
            prefix,
            app=openai_app,
            name="openai",
            middleware=[Middleware(RequestContextMiddleware)],
        )
    )
    self._openai_app = openai_app

    logger.info("server.api_bound", extra={"prefix": prefix})

    return self

citation_slice(annotation, content)

103
104
105
106
107
108
109
110
111

def citation_slice(annotation: Annotation, content: str) -> slice:
    """Convert an OpenAI inclusive citation span to a validated Python slice."""
    citation = annotation.url_citation
    start = citation.start_index
    stop = citation.end_index + 1
    if not 0 <= start < stop <= len(content):
        msg = "citation indices must refer to the final assistant text"
        raise ValueError(msg)
    return slice(start, stop)

client_event(event_type, data, *, namespace=())

52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

def client_event(
    event_type: ClientEventType,
    data: JsonValue,
    *,
    namespace: tuple[str, ...] = (),
) -> dict[str, object]:
    """Build an explicitly public, JSON-safe client stream event."""
    envelope = _ClientEventEnvelope(
        type=_CLIENT_EVENT_ENVELOPE_TYPE,
        schema_version=CLIENT_EVENT_SCHEMA_VERSION,
        event=_ClientEventData(
            type=event_type,
            namespace=namespace,
            data=data,
        ),
    )
    return envelope.model_dump(mode="json")

status_event(description, *, done=False, hidden=False, namespace=())

71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

def status_event(
    description: str,
    *,
    done: bool = False,
    hidden: bool = False,
    namespace: tuple[str, ...] = (),
) -> dict[str, object]:
    """Build a portable status update for native client UI."""
    data = _StatusEventData(
        description=description,
        done=done,
        hidden=hidden,
    )
    return client_event(
        "status",
        data.model_dump(mode="json"),
        namespace=namespace,
    )

checkpoint_scope_dependency(request)

26
27
28
29
30
31

async def checkpoint_scope_dependency(request: Request) -> str:
    """Resolve checkpoint scope."""
    value = request.app.state.checkpoint_scope(request)
    if inspect.isawaitable(value):
        value = await value
    return value

stream_owner_dependency()

11
12
13
14
15
16
17
18
19
20
21
22
23

async def stream_owner_dependency() -> AsyncIterator[_StreamOwner]:
    """
    Manage stream ownership.

    Yields:
        The stream owner dependency instance.

    """
    owner = _StreamOwner()
    try:
        yield owner
    finally:
        await owner.aclose()

reject_legacy_fields(data)

149
150
151
152
153
154
155
156
157
158
159
160

@model_validator(mode="before")
@classmethod
def reject_legacy_fields(cls, data: Any) -> Any:
    """Reject deprecated Chat Completions function parameters."""
    return _reject_legacy_fields(
        data,
        {
            "function_call": "tool_choice",
            "functions": "tools",
        },
        title=cls.__name__,
    )

validate_stream_options()

162
163
164
165
166
167
168

@model_validator(mode="after")
def validate_stream_options(self) -> "ChatCompletionRequest":
    """Allow stream options only for streaming requests."""
    if self.stream_options is not None and not self.stream:
        msg = "stream_options may only be set when stream is true"
        raise ValueError(msg)
    return self

reject_legacy_fields(data)

 94
 95
 96
 97
 98
 99
100
101
102

@model_validator(mode="before")
@classmethod
def reject_legacy_fields(cls, data: Any) -> Any:
    """Reject the deprecated singular Chat Completions function call."""
    return _reject_legacy_fields(
        data,
        {"function_call": "tool_calls"},
        title=cls.__name__,
    )

generate_completion(chat_request, run)

33
34
35
36
37
38
39
40
41

async def generate_completion(
    chat_request: ChatCompletionRequest, run: GraphRun
) -> ChatCompletionResponse:
    """Generate a chat completion."""
    invocation = await invoke_run(run)
    return chat_completion_response(
        model=chat_request.model,
        completion=invocation.output,
    )

stream_completion(chat_request, run)

 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

async def stream_completion(
    chat_request: ChatCompletionRequest, run: GraphRun
) -> AsyncGenerator[str, None]:
    """
    Stream a chat completion response.

    Yields:
        String chunks representing Server-Sent Events.

    """
    include_usage = bool(
        chat_request.stream_options is not None
        and chat_request.stream_options.include_usage
    )
    response_builder = ChatCompletionStreamResponseBuilder(
        chat_request.model,
        include_usage=include_usage,
    )
    final_message: AIMessage | None = None
    text_parts: list[str] = []
    include_client_events = run.config.supports(
        GraphFeature.CLIENT_EVENTS
    ) and stream_events_requested(chat_request.metadata)

    try:  # ruff: ignore[too-many-nested-blocks, too-many-statements-in-try-clause]
        yield response_builder.role()

        run_stream = stream_run(run)
        # Closing the HTTP response must also close the nested graph stream.
        async with aclosing(run_stream):
            async for event in run_stream:
                if isinstance(event, LangGraphInterruptBatch):
                    yield response_builder.interrupt(event)
                    yield response_builder.finish("tool_calls")
                    yield response_builder.done()
                    return

                if isinstance(event, AIMessage):
                    final_message = event
                    continue

                if not isinstance(event, str):
                    if include_client_events:
                        extension = client_event_extension_from_custom_event(event)
                        if extension is not None:
                            yield response_builder.client_event(extension)
                    continue

                text_parts.append(event)
                yield response_builder.text(event)

        final_message = _require_final_message(final_message)
        for chunk in _final_chunks(
            response_builder,
            final_message,
            streamed_text="".join(text_parts) if text_parts else None,
            include_usage=include_usage,
        ):
            yield chunk

    except Exception:
        logger.exception("chat_completion.stream_failed")
        yield response_builder.error("Internal server error")
        yield response_builder.done()

client_event_extension_from_custom_event(event)

20
21
22
23
24
25
26

def client_event_extension_from_custom_event(
    event: CustomStreamPart,
) -> dict[str, object] | None:
    """Validate an explicitly public event and build its stream extension."""
    # Ignore LangGraph's execution namespace, which contains dynamic task IDs.
    # The public namespace is authored explicitly inside the validated event.
    return client_event_extension(event["data"])

stream_events_requested(metadata)

13
14
15
16
17

def stream_events_requested(metadata: dict[str, str] | None) -> bool:
    """Return whether a request opted into the supported event stream version."""
    return (metadata or {}).get(STREAM_EVENTS_METADATA_KEY) == (
        STREAM_EVENTS_METADATA_VALUE
    )

InterruptResume(run_id, state_token, values)

interrupt_arguments(*, run_id, state_token, payload)

47
48
49
50
51
52
53
54
55
56
57
58
59
60

def interrupt_arguments(
    *,
    run_id: str,
    state_token: str,
    payload: Any,
) -> str:
    """Encode one interrupt without coercing unsupported graph values."""
    return _dump_json(
        {
            "run_id": run_id,
            "state_token": state_token,
            "payload": payload,
        }
    )

interrupt_tool_call_id(interrupt_id)

42
43
44

def interrupt_tool_call_id(interrupt_id: str) -> str:
    """Format interrupt tool call ID."""
    return f"{INTERRUPT_TOOL_CALL_ID_PREFIX}{interrupt_id}"

parse_resume_request(messages)

 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

def parse_resume_request(
    messages: list[ChatCompletionRequestMessage],
) -> InterruptResume | None:
    """
    Parse the trailing canonical assistant/tool interrupt exchange.

    Ordinary tool messages remain ordinary graph input. A LangGraph resume is
    recognized only when the tool results answer a preceding assistant message
    whose function calls are all ``langgraph_interrupt`` calls.
    """
    tool_start = _trailing_tool_start(messages)
    if tool_start is None:
        return None

    tool_messages = messages[tool_start:]
    assistant = messages[tool_start - 1] if tool_start > 0 else None

    if assistant is None or assistant.role != Role.ASSISTANT:
        if any(_is_interrupt_tool_result(message) for message in tool_messages):
            msg = "Interrupt tool results must follow their assistant tool calls."
            raise InvalidResumeRequestError(msg)
        return None

    calls = assistant.tool_calls or []
    interrupt_calls = [
        call for call in calls if call.function.name == INTERRUPT_TOOL_NAME
    ]

    if not interrupt_calls:
        if any(_is_interrupt_tool_result(message) for message in tool_messages):
            msg = "Interrupt tool results must follow their assistant tool calls."
            raise InvalidResumeRequestError(msg)
        return None

    if len(interrupt_calls) != len(calls):
        msg = "Interrupt and ordinary tool calls cannot be resumed in one exchange."
        raise InvalidResumeRequestError(msg)

    parsed_calls = _parse_interrupt_calls(interrupt_calls)
    values = _parse_tool_results(tool_messages, parsed_calls)

    run_ids = {call.run_id for call in parsed_calls.values()}
    state_tokens = {call.state_token for call in parsed_calls.values()}
    if len(run_ids) != 1 or len(state_tokens) != 1:
        msg = (
            "All interrupt tool calls in one exchange must belong to the same run "
            "and interrupt generation."
        )
        raise InvalidResumeRequestError(msg)

    return InterruptResume(
        run_id=run_ids.pop(),
        state_token=state_tokens.pop(),
        values=values,
    )

validate_interrupt_payload(payload)

63
64
65

def validate_interrupt_payload(payload: Any) -> None:
    """Reject graph values that cannot cross the OpenAI JSON boundary."""
    _dump_json(payload)

ChatCompletionStreamResponseBuilder(model, *, include_usage=False)

189
190
191
192
193

def __init__(self, model: str, *, include_usage: bool = False) -> None:
    self.response_id = f"chatcmpl-{uuid.uuid4()}"
    self.created = int(time.time())
    self.model = model
    self.include_usage = include_usage

client_event(extension)

203
204
205
206
207
208

def client_event(self, extension: dict[str, object]) -> str:
    """Build an empty-delta chunk carrying the opt-in event extension."""
    return self._chunk(
        ChatCompletionStreamResponseDelta(),
        client_event_extension=extension,
    )

done()

267
268
269

def done(self) -> str:  # ruff: ignore[no-self-use]
    """Stream done."""
    return "data: [DONE]\n\n"

error(message)

261
262
263
264
265

def error(self, message: str) -> str:
    """Stream error."""
    return self._format_data(
        openai_error_payload(ErrorObject(message=message, type="server_error"))
    )

finish(finish_reason, *, annotations=None)

248
249
250
251
252
253
254
255
256
257
258
259

def finish(
    self,
    finish_reason: str,
    *,
    annotations: list[Annotation] | None = None,
) -> str:
    """Stream finish."""
    return self._chunk(
        ChatCompletionStreamResponseDelta(),
        finish_reason=finish_reason,
        annotations=annotations,
    )

interrupt(batch)

210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227

def interrupt(self, batch: LangGraphInterruptBatch) -> str:
    """Stream interrupt."""
    return self._chunk(
        ChatCompletionStreamResponseDelta(
            tool_calls=[
                ChatCompletionStreamToolCall(
                    index=index,
                    id=interrupt_tool_call_id(interrupt.id),
                    type="function",
                    function=ChatCompletionStreamToolCallFunction(
                        name=INTERRUPT_TOOL_NAME,
                        arguments=interrupt_tool_arguments(batch, interrupt),
                    ),
                )
                for index, interrupt in enumerate(batch.interrupts)
            ],
        ),
    )

role()

195
196
197

def role(self) -> str:
    """Stream role."""
    return self._chunk(ChatCompletionStreamResponseDelta(role=Role.ASSISTANT))

text(content)

199
200
201

def text(self, content: str) -> str:
    """Stream text content."""
    return self._chunk(ChatCompletionStreamResponseDelta(content=content))

tool_calls(message)

229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246

def tool_calls(self, message: AIMessage) -> str:
    """Stream complete final-message tool calls as one delta."""
    return self._chunk(
        ChatCompletionStreamResponseDelta(
            tool_calls=[
                ChatCompletionStreamToolCall(
                    index=index,
                    id=tool_call.id,
                    type=tool_call.type,
                    function=ChatCompletionStreamToolCallFunction(
                        name=tool_call.function.name,
                        arguments=tool_call.function.arguments,
                    ),
                )
                for index, tool_call in enumerate(tool_calls_from_message(message))
            ]
        )
    )

usage(usage)

271
272
273
274
275
276
277
278
279
280

def usage(self, usage: UsageMetadata) -> str:
    """Stream the optional final usage-only chunk."""
    response = ChatCompletionStreamResponse(
        id=self.response_id,
        created=self.created,
        model=self.model,
        choices=[],
        usage=usage_info(usage),
    )
    return self._format_data(response.model_dump(mode="json", exclude_none=True))

annotations_from_message(message)

113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

def annotations_from_message(message: AIMessage) -> list[Annotation]:
    """Convert native LangChain citations to OpenAI URL annotations."""
    annotations = []
    text_offset = 0
    for block in message.content_blocks:
        if block["type"] != "text":
            continue
        for citation in block.get("annotations", []):
            if citation.get("type") != "citation":
                continue
            citation = cast("Citation", citation)
            required = {"url", "title", "start_index", "end_index"}
            if not required.issubset(citation):
                continue
            annotation = Annotation.model_validate(
                {
                    "type": "url_citation",
                    "url_citation": {
                        "url": citation["url"],
                        "title": citation["title"],
                        "start_index": citation["start_index"] + text_offset,
                        "end_index": citation["end_index"] + text_offset,
                    },
                }
            )
            span = citation_slice(annotation, message.text)
            cited_text = citation.get("cited_text")
            if cited_text is not None and message.text[span] != cited_text:
                msg = "citation indices must match cited_text"
                raise ValueError(msg)
            annotations.append(annotation)
        text_offset += len(block["text"])
    return annotations