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, safety_identifier and its deprecated alias 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.

Foreground LGOS Responses are not persisted for retrieval or deletion. Omitted, null, and false store values are accepted, and the returned foreground Response reports store=false; foreground store=true is rejected. For a model declaring GraphFeature.BACKGROUND, background=true selects the polling-only path and permits either store=false or store=true; the background engine keeps the result either way, and Hatchet keeps it for its data retention period (30 days by default when self-hosted). Background streaming and cursor/event replay are rejected. conversation is unsupported in both modes. 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:

The lgos command adds the server 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(graphs={...}, run_coordinator=...) to map OpenAI model names to GraphConfig values; registry.graphs is a plain dict. The registry must contain at least one graph. Model IDs must be non-empty, must not contain /, and must not be . or ... registry.get_graph(model) returns a GraphConfig or raises a 404 model_not_found error. run_coordinator is shared by every interrupt-enabled graph and is required only when a graph declares GraphFeature.INTERRUPTS.

LanggraphOpenaiServe(registry=registry, app=app) serves the registry. LanggraphOpenaiServe(..., background=backend) accepts an optional BackgroundBackend. LanggraphOpenaiServe(..., checkpoint_scope=resolver) accepts an optional sync or async callable from FastAPI Request to a non-empty, server-trusted string. Interrupt checkpoint keys and background Response authorization include this scope before model and run identity. Use an authenticated tenant or principal identifier when runs must be isolated between security domains; do not derive the scope from untrusted OpenAI metadata or the end-user safety_identifier and user fields. 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.

RequestContextFilter adds the active request's LGOS fields, such as request_id, to log records; install it on a host handler to enrich every record. See Production Logging.

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: allowlist of server-executed tool names, advertised as lgos.server_tools. 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. See Server Tools.
• client_tools: names of client-owned functions the graph can call, such as display_file, advertised as lgos.client_tools so clients offer only the functions a graph uses. LGOS does not enforce the list, and a name cannot also be a server tool.
• runtime_callbacks: callbacks included in the LangGraph RunnableConfig. When Langfuse tracing is enabled, LGOS adds its callback without mutating this collection or 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, server_tools, and client_tools as frozen sets, so later mutations of the input collections cannot change a registered model. 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. To keep a private model call out of the stream, tag it with LangGraph's nostream tag (langgraph.constants.TAG_NOSTREAM), for example ChatOpenAI(..., tags=[TAG_NOSTREAM]); the call still runs and returns its output, but LangGraph emits none of its tokens. disable_streaming=True is for models that cannot stream: it makes the provider request non-streaming, and LangGraph emits the call's complete message instead. With either option, LangGraph also emits a complete message that a node returns into state. LGOS forwards neither complete message, but other consumers of the graph's stream see them.

A directly supplied compiled graph is reused and validated when its GraphConfig is built. 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 it before execution. Validation checks the context schema and interrupt checkpointer capabilities, rejects static breakpoints (interrupt_before or interrupt_after) in the graph or its subgraph nodes, since LGOS pauses only at interrupt(), and rejects a checkpointer, including checkpointer=True, on a graph without GraphFeature.INTERRUPTS, since clients send the full conversation with every other request. A registry with an interrupt-enabled graph and no run_coordinator fails during GraphRegistry construction. A graph may declare both interrupts and background; a background run that reaches an interrupt completes with lgos_interrupt function calls, and an answer in either mode continues it.

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 (the end user from safety_identifier, or the deprecated user field), 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 run 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. A self-hosted Langfuse must be a release that Langfuse's compatibility matrix lists for the Python SDK version that the extra installs. 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. Before that, LGOS creates the Langfuse client with a tracer provider of its own, as Langfuse recommends beside another tracing backend. Langfuse's observations, which carry prompts and outputs, then never reach the application's OpenTelemetry exporters, and the application's spans never reach Langfuse; both share each trace ID. An application that creates its own Langfuse client first keeps its own tracing setup. 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.

LGOS gives every 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 it names an end user with safety_identifier or the deprecated user) that end user as lgos.user. Graph code can read them with LangGraph's get_config(), with or without Langfuse. When the toggle is on, LGOS also sets the end user, metadata.conversation_id, and the registered model as Langfuse's native user, session, and trace name on every observation of the run; a custom handler passed with the toggle off does not receive them. 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 lgos.features extension and enables server behavior where applicable. 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. GraphFeature.BACKGROUND enables and advertises polling-only background Responses.

OpenTelemetry¶

LGOS reports every graph execution through the OpenTelemetry API, following the GenAI workflow span and workflow metric conventions. The API records nothing until the host application configures an SDK, for example with opentelemetry-instrument.

HTTP request telemetry belongs to the host app. The mounted OpenAI app turns off FastAPI's native OpenTelemetry, so the host's telemetry settings, including turning it off, cover LGOS routes, and http.route includes the mount, such as /v1/responses.

gen_ai.workflow.name is the registered model. Both signals cover graph execution through the final output; they exclude lease waits, background queue time, and checkpoint cleanup. While LGOS advances the graph, the span's context is current: spans the graph creates, such as model calls, are its children, while code that consumes a stream keeps its own parent span between events. The span itself is not current, so code in the graph cannot change it, and it carries only the attributes above; add attributes to spans of your own. A run that ends before its output exists fails: one that raises an exception, one cancelled by a disconnecting client or a cancelled background Response, and one whose stream the consumer closes mid-run. The span status is then ERROR with the exception message, if any, and error.type is the exception type, qualified by its module unless it is built in, such as asyncio.exceptions.CancelledError or GeneratorExit. Closing a stream after its final output is not a failure. Design Choices lists the tracers whose cancellation handling this follows. A raised Exception propagates to its handler: LGOS routes and the background engines log it with its traceback, and direct Python callers receive it; cancellations and closed streams are not logged. The span does not repeat an exception as an event. The histogram uses the conventions' bucket boundaries, which start at one second; configure an SDK View for finer boundaries.

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 deterministic default, because clients omit values equal to the advertised defaults; a default_factory must return the same value on every call. Registration fails when the model cannot build its defaults or schema.

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 the settings with the model's strict JSON validation 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 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.

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(), aput(), aput_writes(), and adelete_thread();
• configure an asynchronous GraphRegistry.run_coordinator; and
• use a durable checkpointer and cross-process coordinator in production.

The coordinator provides single-flight leases for interrupt runs: it returns an async context manager and rejects an occupied checkpoint key instead of queueing it. Exiting that context manager must release the lease even when the exit is cancelled.

The initial request does not require metadata. LGOS generates a UUID run ID for every new run and embeds it in the paused Response ID. InMemoryRunCoordinator is suitable only for tests and a single-process development server; it cannot serialize requests across workers or hosts.

Expire Paused Runs¶

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. A run abandoned after its batch is returned stays until deleted: call delete_expired_interrupt_runs(checkpointer, run_coordinator, older_than=...) from langgraph_openai_serve.graph.interrupt on a schedule, as LangGraph Agent Server's checkpointer TTL does. In production, prefer one scheduled job over a loop in every replica, for example a Kubernetes CronJob with concurrencyPolicy: Forbid or your task queue's scheduler. It deletes the runs whose latest pause is older than older_than and returns their count. It reads every checkpoint through alist(), leaves threads LGOS did not create alone, and skips runs whose lease is held. Choose a TTL longer than the longest time a user may take to answer. To write your own cleanup, select threads whose checkpoint metadata contains OPERATION_ID_METADATA_KEY from the same module, then hold each run's lease, confirm that its latest checkpoint in any namespace is still older than your TTL, and delete it through the checkpointer. lgos serve runs this sweep in each API process, configured by LGOS_INTERRUPT_TTL_MINUTES and LGOS_INTERRUPT_SWEEP_INTERVAL_MINUTES; set the interval to 0 when one scheduled job sweeps instead.

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() before serving requests, serializing migration attempts when workers can start together. Their migrations use CREATE INDEX CONCURRENTLY, which deadlocks with a worker blocked in pg_advisory_lock; retry pg_try_advisory_lock instead. 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. Each process applies pending LangGraph migrations during startup under a separate schema advisory lock. Busy interrupt leases fail before streaming begins with HTTP 409 and code: "run_busy".

Background Execution¶

LanggraphOpenaiServe(background=...) accepts any BackgroundBackend: an engine that runs BackgroundJobs and stores their status and result.

The engine's worker calls execute_background_job(job, run_id, registry). It runs the job through the foreground Responses path and returns the Response JSON the engine stores. Run IDs must be UUIDs because the public Response ID embeds the run ID, so reading a Response needs no lookup table. InMemoryBackgroundBackend(registry) runs jobs as tasks of one process for development and tests; enter its lifespan in the application's lifespan.

Install langgraph-openai-serve[hatchet] for langgraph_openai_serve.integrations.hatchet. create_hatchet_task(hatchet) registers the task in the API and worker processes, with a 30-minute schedule_timeout and a one-hour execution_timeout by default. HatchetBackgroundBackend(task, hatchet.runs) submits, reads, and cancels its runs. The worker's Hatchet lifespan yields the GraphRegistry, including its run_coordinator, that the task executes. The task has no retries, and its 24-hour Hatchet idempotency key is the scoped Idempotency-Key digest. The adapter is never imported by the core package.

Background create accepts an optional Idempotency-Key header of 1 to 255 characters. A retry returns the original Response without starting a second run; reuse with different content returns 422 with code="idempotency_key_reused". A background interrupt run chooses its run ID at submission.

See Run Responses In The Background for the client contract, graph requirements, and deployment wiring.

Server¶

Install langgraph-openai-serve[server] for the lgos command and langgraph_openai_serve.server. lgos serve module:attribute and lgos worker module:attribute run a RegistryFactory: a callable that takes ServerResources (checkpointer, store, run_coordinator) and returns a GraphRegistry, or an async context manager that yields one. lgos serve also accepts every Uvicorn option and UVICORN_* variable. create_app(factory, settings=...) returns the FastAPI application that lgos serve runs; its lifespan opens the resources and mounts the OpenAI routes. ServerSettings reads the LGOS_* server settings. See Run The LGOS Server.

Streaming Status¶

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

# Perform the long-running work.

writer(status_event("Audio ready"))

The helper returns this custom stream value:

{
  "type": "lgos.status",
  "description": "Generating audio"
}

Status text is deliberately authored by the graph; LGOS does not infer it from internal node names or state. Graphs need no feature declaration to publish status.

Status is streaming-only. Streaming Responses needs no metadata opt-in and emits each non-empty description as a standard phase="commentary" message. Non-streaming Responses and Chat Completions ignore statuses. Both APIs ignore any other custom stream data. Use standard Responses function calls plus the Files API for portable durable rich output.

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.

cancel(run_id)

87
88
89

async def cancel(self, run_id: str) -> None:
    """Stop one run; a finished run keeps its outcome."""
    ...

get(run_id)

83
84
85

async def get(self, run_id: str) -> BackgroundRun | None:
    """Read one run, or return None when the engine does not have it."""
    ...

submit(job)

79
80
81

async def submit(self, job: BackgroundJob) -> BackgroundRun:
    """Start ``job``, or return the run already holding its idempotency key."""
    ...

response_id

snapshot()

66
67
68
69
70
71
72
73

def snapshot(self) -> Response:
    """Return the public Response for the run's current state."""
    if self.response is not None:
        return Response.model_validate(self.response)
    response = _queued_response(self.job, self.response_id)
    if self.status in {"completed", "failed"}:
        return _failed_response(response, "The background response failed.")
    return response.model_copy(update={"status": self.status})

ClientFunctionTool(name, description, parameters, strict)

default_values()

54
55
56
57
58

@classmethod
def default_values(cls) -> dict[str, JsonValue]:
    """Return the JSON defaults advertised by model discovery."""
    # Aliases are the public names in the schema and request validation.
    return cls().model_dump(mode="json", by_alias=True)

json_schema()

49
50
51
52

@classmethod
def json_schema(cls) -> dict[str, JsonValue]:
    """Return the JSON Schema advertised by model discovery."""
    return {"$schema": JSON_SCHEMA_DIALECT, **cls.model_json_schema()}

validate_request(request)

32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

@classmethod
def validate_request(cls, request: GraphRequest) -> Self:
    """Read and validate this model's values from an OpenAI request."""
    try:
        return cls.model_validate_json(
            request.metadata.get(SETTINGS_METADATA_KEY, "{}")
        )
    except ValidationError as exc:
        error = exc.errors()[0]
        field = ".".join(str(part) for part in error["loc"])
        label = f"runtime setting for {field}" if field else "runtime settings"
        msg = f"Invalid {label}: {error['msg']}"
        raise InvalidRequestError(
            msg,
            param=f"metadata.{SETTINGS_METADATA_KEY}",
        ) from exc

build_context(request, graph)

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

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

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

build_input(request, messages)

102
103
104
105
106
107
108
109
110

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)

138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

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 GraphError(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 GraphError(msg)
    return message

resolve_graph()

 96
 97
 98
 99
100

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

supports(feature)

92
93
94

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

validate_client_settings(value)

64
65
66
67
68
69
70
71
72
73
74

@field_validator("client_settings")
@classmethod
def validate_client_settings(
    cls,
    value: type[ClientSettings] | None,
) -> type[ClientSettings] | None:
    """Fail at registration when a settings model cannot be advertised."""
    if value is not None:
        value.json_schema()
        value.default_values()
    return value

validate_compiled_graph()

85
86
87
88
89
90

@model_validator(mode="after")
def validate_compiled_graph(self) -> Self:
    """Fail at registration when a directly supplied graph cannot be served."""
    if isinstance(self.graph, CompiledStateGraph):
        _validate_resolved_graph(self.graph, self)
    return self

validate_tool_ownership()

76
77
78
79
80
81
82
83

@model_validator(mode="after")
def validate_tool_ownership(self) -> Self:
    """Keep the side that executes each advertised tool unambiguous."""
    if shared := self.server_tools & self.client_tools:
        names = ", ".join(sorted(shared))
        msg = f"Tools cannot be both server and client tools: {names}"
        raise ValueError(msg)
    return self

GraphRegistry(graphs, run_coordinator=None)

__post_init__()

249
250
251
252
253
254
255
256
257
258
259
260
261
262
263

def __post_init__(self) -> None:
    """Fail at startup for a registry the OpenAI routes cannot serve."""
    if not self.graphs:
        msg = "GraphRegistry must contain at least one graph."
        raise ValueError(msg)
    for model_id in self.graphs:
        # Each model must be addressable as GET /models/{model}.
        if not model_id or "/" in model_id or model_id in {".", ".."}:
            msg = f"Model ID {model_id!r} is not addressable."
            raise ValueError(msg)
    if self.run_coordinator is None and any(
        config.supports(GraphFeature.INTERRUPTS) for config in self.graphs.values()
    ):
        msg = "Interrupt-enabled graphs need a GraphRegistry run_coordinator."
        raise ValueError(msg)

get_graph(model)

265
266
267
268
269
270
271
272
273
274
275
276

def get_graph(self, model: str) -> GraphConfig:
    """Return the graph served as ``model``."""
    try:
        return self.graphs[model]
    except KeyError as exc:
        msg = f"The model '{model}' does not exist."
        raise InvalidRequestError(
            msg,
            param="model",
            code="model_not_found",
            status_code=404,
        ) from exc

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

InMemoryBackgroundBackend(registry)

129
130
131
132
133
134

def __init__(self, registry: GraphRegistry) -> None:
    self._registry = registry
    self._runs: dict[str, BackgroundRun] = {}
    self._keys: dict[str, str] = {}
    self._scopes: dict[str, CancelScope] = {}
    self._tasks: TaskGroup | None = None

cancel(run_id)

165
166
167
168
169
170

async def cancel(self, run_id: str) -> None:
    """Mark an active run cancelled and cancel its task."""
    if run_id in self._runs:
        self._transition(run_id, "cancelled")
    if (scope := self._scopes.get(run_id)) is not None:
        scope.cancel()

get(run_id)

161
162
163

async def get(self, run_id: str) -> BackgroundRun | None:
    """Read one run."""
    return self._runs.get(run_id)

lifespan(_app)

136
137
138
139
140
141
142
143
144
145

@asynccontextmanager
async def lifespan(self, _app: object) -> AsyncGenerator[None, None]:
    """Own the execution tasks for one ASGI application lifespan."""
    async with create_task_group() as tasks:
        self._tasks = tasks
        try:
            yield
        finally:
            self._tasks = None
            tasks.cancel_scope.cancel()

submit(job)

147
148
149
150
151
152
153
154
155
156
157
158
159

async def submit(self, job: BackgroundJob) -> BackgroundRun:
    """Start one task, or return the run holding the job's key."""
    if self._tasks is None:
        msg = "The in-memory background backend lifespan is not running."
        raise RuntimeError(msg)
    if (run_id := self._keys.get(job.idempotency_key)) is not None:
        return self._runs[run_id]
    run = BackgroundRun(id=str(uuid.uuid4()), job=job, status="queued")
    self._runs[run.id] = run
    self._keys[job.idempotency_key] = run.id
    self._scopes[run.id] = CancelScope()
    self._tasks.start_soon(self._execute, run)
    return run

InvalidRequestError(
    message, *, param=None, code=None, status_code=status.HTTP_400_BAD_REQUEST
)

20
21
22
23
24
25
26
27
28
29
30
31

def __init__(
    self,
    message: str,
    *,
    param: str | None = None,
    code: str | None = None,
    status_code: int = status.HTTP_400_BAD_REQUEST,
) -> None:
    super().__init__(message)
    self.param = param
    self.code = code
    self.status_code = status_code

LanggraphOpenaiServe(registry, app=None, checkpoint_scope=None, background=None)

 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

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

    Args:
        registry: The graphs to serve and their interrupt run coordinator.
        app: The host FastAPI application to mount the OpenAI API on. If None,
            a new FastAPI app will be created.
        checkpoint_scope: Optional server-trusted resolver used to isolate
            interrupt checkpoints by deployment or authenticated principal.
        background: Optional polling-only background backend.

    """
    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.background = background
    self.graph_registry = registry

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

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

openai_app

bind_openai_api(prefix=None)

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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

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(),
        # FastAPI's native telemetry belongs to the host app, so the host's
        # settings, including turning it off, also cover the mounted routes.
        telemetry={"tracing": False, "metrics": False, "logs": False},
        **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
    openai_app.state.background_backend = self.background
    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)

RequestContextFilter()

68
69
70

def __init__(self) -> None:
    super().__init__()
    self._log_context = _log_context

filter(record)

72
73
74
75
76
77

def filter(self, record: logging.LogRecord) -> bool:
    """Enrich a record while preserving fields supplied by the caller."""
    for name, value in (self._log_context.get() or {}).items():
        if name not in record.__dict__:
            setattr(record, name, value)
    return True

citation_slice(start_index, end_index, content)

11
12
13
14
15
16
17

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 GraphError(msg)
    return slice(start_index, stop)

execute_background_job(job, engine_run_id, registry)

 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

async def execute_background_job(
    job: BackgroundJob,
    engine_run_id: str,
    registry: GraphRegistry,
) -> dict[str, JsonValue]:
    """Execute one job through the foreground Responses path."""
    request = ResponseCreateRequest.model_validate(job.request)
    response_id = interrupt_response_id(job.run_id, engine_run_id)
    try:
        run = await prepare_response_run(
            request,
            registry,
            checkpoint_scope=job.owner_scope,
            run_id=job.run_id,
        )
        response = await collect_response(
            request,
            run,
            response_id=response_id,
            created_at=job.created_at,
        )
    except InvalidRequestError as exc:
        # The foreground path reports these as 4xx; a background Response
        # reports them as its own failure. An interrupt answer, for example,
        # is validated only once its run holds the lease, so a stale answer
        # fails here.
        response = _failed_response(_queued_response(job, response_id), str(exc))
    return response.model_dump(mode="json", by_alias=True)

status_event(description)

6
7
8

def status_event(description: str) -> dict[str, str]:
    """Build a status update for a node to write with ``get_stream_writer()``."""
    return {"type": STATUS_EVENT_TYPE, "description": description}

convert_to_lc_messages(messages)

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

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 InvalidRequestError(msg, param="messages")
                lc_messages.append(
                    ToolMessage(
                        content=_langchain_content(m.content),
                        name=m.name,
                        tool_call_id=m.tool_call_id,
                    )
                )
    return lc_messages

decode_chat_request(request)

18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

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.safety_identifier or 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 GraphError(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()

145
146
147
148
149
150
151

@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