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.

Responses Request¶

The route accepts string or ordered message input, instructions, plain input_text, input_file.file_id, string-valued metadata, user, flat client function tools, registered name-only custom-tool selectors, standard web_search, their choices, parallel_tool_calls, plain text output, and streaming. Replayed assistant output messages preserve phase; complete function_call items and matching string-valued function_call_output items support ordinary client-tool continuation. Interrupt continuation sends only matching function_call_output items with previous_response_id. Server execution returns native custom_tool_call and custom_tool_call_output pairs or a web_search_call in the same response; complete output items can also be replayed as history. LGOS executes registered custom tools inside that response. This intentionally differs from OpenAI's ordinary custom-tool flow, where caller code executes the tool and supplies its output to a later model request; the wire items remain standard Responses types. The public web_search shape does not prescribe the graph's search backend; the bundled demo chooses an HTTP or upstream provider backend.

LGOS does not persist completed Responses for retrieve or deletion. Omitted, null, and false store values are accepted, and the returned Response reports store=false; store=true, conversation, and background mode are rejected. previous_response_id is supported for interruptible graphs to resume execution (and rejected for non-interruptible graphs); new instructions are rejected on those resumes. The route also rejects unregistered custom tools, client-supplied custom descriptions or formats, other built-in tools, structured output, image/audio input, URL or inline file input, function result-content lists, reasoning and generation controls, include, stream options, service tiers, reusable prompts, prompt-cache controls, and truncation. Unknown fields are not silently ignored. See the supported Responses subset for the complete behavior and continuation rules.

Chat Completions Request¶

Chat messages accept string content, explicit text content parts, and native file parts containing only file.file_id. The route supports modern function tools, tool_choice, assistant tool_calls, matching tool messages, streaming, and stream_options.include_usage. Image and audio parts, inline file data or filenames, prompt-cache fields, deprecated function fields, generation controls, and other unknown fields are rejected.

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 copies its initial mapping and must contain at least one graph. It rejects empty model IDs, ., .., and IDs containing /. The public registry.registry mapping is an insertion-ordered, read-only view; use registry.register(model_id, config) to add or replace a graph. Replacing an existing ID preserves its position.

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 OpenAI metadata or the OpenAI 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.

Responses input_file.file_id content and native Chat file parts normalize to the same LangChain file block, so graphs receive native file_id values and decide whether to download, parse, or forward them. File upload and storage belong to an external OpenAI Files API, not the LGOS package. See Accept And Display Files.

GraphConfig accepts:

• graph: compiled graph, sync factory, or async factory.
• description: required human-readable model description advertised by model listing and retrieval.
• features: GraphFeature values that enable optional server behavior or advertise graph input and client-tool capabilities.
• client_settings: explicit public ClientSettings model class advertised by model retrieval.
• server_tools: internal allowlist of server-executed tool names. A registered custom tool is selected with the Responses custom type and name; web_search uses its built-in type. The graph owns tool definitions and execution. Model retrieval does not advertise tools. See Server Tools.
• 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 normalized request and LangChain messages to graph input.
• context_factory(request, client_settings): compose the final typed LangGraph runtime context from normalized request values, server-owned values, and optional validated public settings.
• output_to_message(output): custom graph output to a durable AIMessage.

GraphConfig is immutable after construction. Pydantic snapshots features and server_tools as frozen sets, so later mutations of the input collections cannot change a registered model. To change a declaration, construct a replacement and pass it to registry.register(). Freezing the declaration does not make a caller-owned callback handler or callback manager internally immutable.

Streaming forwards non-empty text from every AIMessageChunk emitted by the graph's messages stream. Configure private ChatOpenAI calls with disable_streaming=True; LangChain then uses the complete invocation path and does not emit model stream chunks for that call.

A directly supplied compiled graph is reused. A sync or async graph factory is called for every request and is never cached; LGOS validates each resolved value as a compiled state graph and rechecks its context schema and interrupt checkpointer capabilities before execution. Static configuration relationships, including the requirement that run_coordinator appear exactly when GraphFeature.INTERRUPTS is enabled, fail during GraphConfig construction.

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].

Graph adapters receive an immutable, protocol-neutral GraphRequest from either API's decoder. It exposes the shared model, metadata, user, normalized client function tools, selected server-tool names in server_tools, tool_choice, and parallel_tool_calls values. NamedFunctionToolChoice identifies a required client function, while NamedCustomToolChoice identifies a required registered custom tool. A single web_search declaration with tool_choice="required" requires search. Named built-in choices are outside the supported subset. Raw OpenAI transport models are not part of the graph-adapter interface.

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 Tracing¶

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.graph_run for both endpoints and adds RunnableConfig.metadata fields for the request ID, registered graph model, (for interrupt runs) operation ID, and (when the request supplies metadata.conversation_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 features set is returned in the versioned lgos.features extension and enables server behavior where applicable. GraphFeature.CLIENT_EVENTS enables and advertises public status commentary in streaming Responses. Chat Completions ignores custom stream events and does not emit commentary. GraphFeature.MCP_TOOLS advertises that a client may attach and execute tools from its configured MCP gateway; it does not publish tool definitions or grant access to them. GraphFeature.FILE_INPUTS advertises that the graph resolves native file content parts. 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.lgos_settings string. Clients omit values equal to the advertised defaults. System instructions remain ordinary OpenAI messages and are independent of ClientSettings; native OpenAI 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 lgos.client_settings, with independent schema_version, json_schema, and defaults fields. All client settings use the fixed metadata.lgos_settings key. Clients use the descriptor's validated defaults object as the baseline; default keywords within the generated JSON Schema are annotations, not the runtime baseline. The schema's $schema keyword declares the JSON Schema 2020-12 dialect independently of the LGOS descriptor version.

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 embeds it in the paused Response ID. A caller that needs deterministic initial-request retries can instead supply a non-nil UUID in metadata.lgos_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".

Streaming Status¶

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 helper writes this versioned graph-to-LGOS envelope:

{
  "type": "lgos.client_event",
  "schema_version": 1,
  "event": {
    "type": "status",
    "namespace": ["media"],
    "data": {
      "description": "Generating audio",
      "done": false,
      "hidden": false
    }
  }
}

Status text is deliberately authored by the graph; LGOS does not infer it from internal node names or state. Responses exposes the description as commentary and suppresses hidden updates; the namespace, done, and hidden fields do not become nonstandard Response fields.

The event envelope has its own schema version, independent of model discovery and client settings. The v1 event vocabulary is status, progress, and artifact. client_event("status", data) remains the lower-level equivalent when an application already has validated status data; prefer status_event() for its typed fields. Event data must be JSON-safe, and every namespace segment must be a string. The namespace is a stable, author-defined path; LGOS does not expose LangGraph's dynamic execution namespace.

Status is streaming-only and always requires the graph feature. Responses needs no metadata opt-in and emits each visible update as a standard phase="commentary" message. The Chat Completions API is strictly for simple graphs and plain text streaming; it ignores custom stream events and does not emit commentary. Responses ignores progress and artifact. Use standard Responses function calls plus the Files API for portable durable rich output. Unknown custom events remain available only to direct runner consumers.

See Streaming status for the wire contract and Stream final text and commentary 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(start_index, end_index, text) to validate them and create a Python slice. Responses maps citations to output_text.annotations and emits the typed annotation event while streaming. Chat maps them to completed message.annotations; its final streaming delta uses the compatibility extension.

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.

ClientFunctionTool(name, description, parameters, strict)

defaults()

32
33
34
35
36
37
38

@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)

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
71
72

@classmethod
def validate_request(cls, request: GraphRequest) -> Self:
    """Read and validate this model's values from an OpenAI request."""
    parameter = f"metadata.{SETTINGS_METADATA_KEY}"
    encoded = request.metadata.get(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)

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

async def build_context(
    self,
    request: GraphRequest,
    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)

121
122
123
124
125
126
127
128
129

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

render_output(output)

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

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()

113
114
115
116
117
118
119

async def resolve_graph(self) -> CompiledStateGraph:
    """Get the graph instance, resolving callable graph factories."""
    if isinstance(self.graph, CompiledStateGraph):
        graph = self.graph
    else:
        graph = await _maybe_await(self.graph())
    return _validate_resolved_graph(graph, self)

supports(feature)

109
110
111

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

validate_interrupt_configuration()

 97
 98
 99
100
101
102
103
104
105
106
107

@model_validator(mode="after")
def validate_interrupt_configuration(self) -> Self:
    """Validate feature relationships that do not depend on a resolved graph."""
    interrupt_enabled = self.supports(GraphFeature.INTERRUPTS)
    if self.run_coordinator is not None and not interrupt_enabled:
        msg = "run_coordinator is only supported by interrupt-enabled graphs."
        raise ValueError(msg)
    if interrupt_enabled and self.run_coordinator is None:
        msg = "Interrupt-enabled graphs must configure a run_coordinator."
        raise ValueError(msg)
    return self

GraphRegistry(*, registry)

252
253
254
255
256
257
258
259
260
261
262

def __init__(self, *, registry: Mapping[str, GraphConfig]) -> None:
    if not registry:
        msg = "GraphRegistry must contain at least one graph."
        raise ValueError(msg)

    entries = {
        _validate_model_id(model_id): _validate_graph_config(config)
        for model_id, config in registry.items()
    }
    self._entries = entries
    self._registry = MappingProxyType(entries)

registry

get_graph(name)

279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297

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()

275
276
277

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

register(model_id, config)

269
270
271
272
273

def register(self, model_id: str, config: GraphConfig) -> None:
    """Add or replace one graph through the validated registry boundary."""
    validated_model_id = _validate_model_id(model_id)
    validated_config = _validate_graph_config(config)
    self._entries[validated_model_id] = validated_config

GraphRequest(
    model, metadata, user, tools, tool_choice, parallel_tool_calls, server_tools=()
)

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

 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
113

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)

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
164
165

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)
    openai_app.include_router(responses_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

NamedCustomToolChoice(name)

NamedFunctionToolChoice(name)

citation_slice(start_index, end_index, content)

 9
10
11
12
13
14
15

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

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

55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

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_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=())

74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

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,
    )

convert_to_lc_messages(messages)

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
71
72
73
74
75
76
77
78
79
80
81

def convert_to_lc_messages(
    messages: list[ChatCompletionRequestMessage],
) -> list[BaseMessage]:
    """
    Convert OpenAI messages to LangChain messages.

    This function converts a list of OpenAI-compatible message objects to their
    LangChain equivalents for use with LangGraph.

    Args:
        messages: A list of OpenAI chat completion request messages to convert.

    Returns:
        A list of LangChain message objects.

    """
    lc_messages: list[BaseMessage] = []
    for m in messages:
        match m.role:
            case Role.SYSTEM:
                lc_messages.append(
                    SystemMessage(content=_langchain_content(m.content), name=m.name)
                )
            case Role.USER:
                lc_messages.append(
                    HumanMessage(content=_langchain_content(m.content), name=m.name)
                )
            case Role.ASSISTANT:
                lc_messages.append(_assistant_message(m))
            case Role.TOOL:
                if m.tool_call_id is None:
                    msg = "Tool messages require the 'tool_call_id' field."
                    raise InvalidChatMessageError(msg)
                lc_messages.append(
                    ToolMessage(
                        content=_langchain_content(m.content),
                        name=m.name,
                        tool_call_id=m.tool_call_id,
                    )
                )
    return lc_messages

UnsupportedChatRequestError(message, *, param)

21
22
23

def __init__(self, message: str, *, param: str) -> None:
    super().__init__(message)
    self.param = param

decode_chat_request(request)

26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

def decode_chat_request(
    request: ChatCompletionRequest,
) -> tuple[GraphRequest, list[BaseMessage]]:
    """Normalize one Chat Completions request for graph execution."""
    graph_request = GraphRequest(
        model=request.model,
        metadata=dict(request.metadata or {}),
        user=request.user,
        tools=tuple(
            ClientFunctionTool(
                name=tool.function.name,
                description=tool.function.description,
                parameters=(
                    dict(tool.function.parameters)
                    if tool.function.parameters is not None
                    else None
                ),
                strict=tool.function.strict,
            )
            for tool in request.tools or ()
        ),
        tool_choice=_decode_tool_choice(request.tool_choice),
        parallel_tool_calls=request.parallel_tool_calls,
    )
    return (
        graph_request,
        convert_to_lc_messages(request.messages),
    )

ChatCompletionStreamResponseBuilder(model, *, include_usage=False)

121
122
123
124
125

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

done()

173
174
175
176

@staticmethod
def done() -> str:
    """Stream done."""
    return "data: [DONE]\n\n"

error(message)

167
168
169
170
171

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)

154
155
156
157
158
159
160
161
162
163
164
165

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

role()

127
128
129

def role(self) -> str:
    """Stream role."""
    return self._chunk(ChoiceDelta(role="assistant"))

text(content)

131
132
133

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

tool_calls(message)

135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

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

usage(usage)

178
179
180
181
182
183
184
185
186
187
188

def usage(self, usage: UsageMetadata) -> str:
    """Stream the optional final usage-only chunk."""
    response = ChatCompletionChunk(
        id=self.response_id,
        object="chat.completion.chunk",
        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)

 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104

def annotations_from_message(message: AIMessage) -> list[Annotation]:
    """Convert validated LangChain citations to Chat URL annotations."""
    return [
        Annotation.model_validate(
            {
                "type": "url_citation",
                "url_citation": {
                    key: citation[key]
                    for key in ("url", "title", "start_index", "end_index")
                },
            }
        )
        for citation in citations_from_message(message)
    ]

chat_completion_response(*, model, message)

29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

def chat_completion_response(
    *,
    model: str,
    message: AIMessage,
) -> ChatCompletion:
    """Build a non-streaming OpenAI-compatible chat completion response."""
    resp_message, finish_reason = response_message(message)
    return ChatCompletion(
        id=f"chatcmpl-{uuid.uuid4()}",
        object="chat.completion",
        created=int(time.time()),
        model=model,
        choices=[
            ChatChoice(
                index=0,
                message=resp_message,
                finish_reason=finish_reason,
            )
        ],
        usage=usage_info(message.usage_metadata),
    )

response_message(message)

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

def response_message(
    message: AIMessage,
) -> tuple[ChatCompletionMessage, Literal["stop", "tool_calls"]]:
    """Format response message."""
    tool_calls = tool_calls_from_message(message)
    return (
        ChatCompletionMessage(
            role="assistant",
            content=message.text or None,
            annotations=annotations_from_message(message) or None,
            tool_calls=tool_calls or None,
        ),
        "tool_calls" if tool_calls else "stop",
    )

tool_calls_from_message(message)

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

def tool_calls_from_message(
    message: AIMessage,
) -> list[ChatCompletionMessageFunctionToolCall]:
    """Convert native LangChain tool calls to Chat Completions tool calls."""
    tool_calls = []
    for tool_call in message.tool_calls:
        tool_call_id = tool_call.get("id")
        if not tool_call_id:
            msg = "Final AIMessage tool calls must have an id."
            raise ValueError(msg)
        tool_calls.append(
            ChatCompletionMessageFunctionToolCall(
                id=tool_call_id,
                type="function",
                function=Function(
                    name=tool_call["name"],
                    arguments=json.dumps(tool_call["args"]),
                ),
            )
        )
    return tool_calls

usage_info(usage)

107
108
109
110
111
112
113
114
115

def usage_info(usage: UsageMetadata | None) -> CompletionUsage | None:
    """Map LangChain's provider-reported usage to Chat Completions usage."""
    if usage is None:
        return None
    return CompletionUsage(
        prompt_tokens=usage["input_tokens"],
        completion_tokens=usage["output_tokens"],
        total_tokens=usage["total_tokens"],
    )

validate_stream_options()

142
143
144
145
146
147
148

@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

generate_completion(chat_request, run)

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

async def generate_completion(
    chat_request: ChatCompletionRequest, run: GraphRun
) -> ChatCompletion:
    """Generate a chat completion."""
    async with run:
        output = await invoke_run(run)
        if not isinstance(output, AIMessage):
            msg = "The graph returned an unsupported Chat Completions output."
            raise TypeError(msg)
        return chat_completion_response(
            model=chat_request.model,
            message=output,
        )

prepare_completion_run(request, graph_registry)

31
32
33
34
35
36
37
38
39
40
41
42
43
44

async def prepare_completion_run(
    request: ChatCompletionRequest,
    graph_registry: GraphRegistry,
) -> GraphRun:
    """Validate a Chat request and prepare its graph run."""
    graph_request, messages = decode_chat_request(request)
    graph_config = graph_registry.get_graph(request.model)
    if graph_config.supports(GraphFeature.INTERRUPTS):
        message = (
            f"Model '{request.model}' requires interrupts, which is only "
            "supported via the Responses API (/v1/responses)."
        )
        raise UnsupportedChatRequestError(message, param="model")
    return await prepare_run(graph_request, messages, graph_registry)

stream_completion(chat_request, run)

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

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,
    )
    chunks = _generate_stream_chunks(response_builder, run, include_usage=include_usage)
    try:
        async with aclosing(chunks):
            async for chunk in chunks:
                yield chunk
    except Exception:
        logger.exception("chat_completion.stream_failed")
        yield response_builder.error("Internal server error")
        yield response_builder.done()

create_chat_completion(chat_request, graph_registry, stream_owner)

24
25
26
27
28
29
30
31
32
33
34
35
36
37
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

@router.post(
    "/chat/completions",
    response_model=ChatCompletion,
    response_model_exclude_none=True,
)
async def create_chat_completion(
    chat_request: ChatCompletionRequest,
    graph_registry: Annotated[GraphRegistry, Depends(get_graph_registry)],
    stream_owner: Annotated[
        StreamOwner,
        Depends(get_stream_owner, scope="request"),
    ],
) -> StreamingResponse | ChatCompletion:
    """
    Create a chat completion.

    This endpoint is compatible with OpenAI's chat completion API.

    Args:
        chat_request: The parsed chat completion request.
        graph_registry: The graph registry dependency.
        stream_owner: The request-scoped streaming task owner.

    Returns:
        A chat completion response, either as a complete response or as a stream.

    """
    bind_log_context(
        model=chat_request.model,
        stream=chat_request.stream,
    )

    with graph_errors(input_param="messages"):
        try:
            run = await chat_service.prepare_completion_run(
                chat_request,
                graph_registry,
            )
        except (InvalidChatMessageError, UnsupportedChatRequestError) as exc:
            raise OpenAIHTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                error=ErrorObject(
                    message=str(exc),
                    type="invalid_request_error",
                    param=(
                        exc.param
                        if isinstance(exc, UnsupportedChatRequestError)
                        else "messages"
                    ),
                ),
            ) from exc

        if chat_request.stream:
            body = stream_owner.start(
                chat_service.stream_completion(chat_request, run),
                run,
            )
            return StreamingResponse(
                body,
                media_type="text/event-stream",
            )

        return await chat_service.generate_completion(chat_request, run)