Reference¶

OpenAI-Compatible API¶

Default prefix: /v1. Change it with LGOS_OPENAI_API_PREFIX or bind_openai_api(prefix=...).

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

Settings¶

Package settings:

Demo API and graph settings:

Settings for optional clients and gateways are documented under Integrations.

Public API¶

Use LanggraphOpenaiServe to bind OpenAI-compatible routes to a FastAPI app. Use GraphRegistry to map OpenAI model names to GraphConfig values. The registry must contain at least one graph. Pydantic rejects empty registries and model IDs that cannot be addressed as one URL path segment. Registry keys are read-only after validation; use registry.register(model_id, config) to add or replace a graph.

GraphConfig accepts:

• graph: compiled graph, sync factory, or async factory.
• streamable_node_names: node names whose streamed AIMessageChunk values are forwarded to clients.
• features: GraphFeature values that enable optional server behavior.
• client_settings: explicit public ClientSettings model class advertised by model retrieval.
• runtime_callbacks: callbacks included in the LangGraph RunnableConfig.
• request_to_input(request, messages): custom OpenAI request to graph input.
• context_factory(request, client_settings): compose the final typed LangGraph runtime context from server-owned values and optional validated public settings.
• output_to_text(output): custom graph output to assistant text.

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

LGOS assembles runnable config from runtime_callbacks and, when present, the checkpoint 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.

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

Runtime Settings¶

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

from pydantic import Field

from langgraph_openai_serve import ClientSettings


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

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

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

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

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

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

Interrupt-enabled graphs must be compiled with a LangGraph checkpointer and requests must include metadata={"langgraph_thread_id": "<client-chat-id>"}. Use a durable checkpointer in production.

Client Stream Events¶

Inside a graph node or tool, mark a passive client notification as explicitly public with client_event():

from langgraph.config import get_stream_writer
from langgraph_openai_serve import client_event

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

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

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

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

Citation Events¶

Inside a graph node or tool, emit a citation with LangGraph's stream writer:

from langgraph.config import get_stream_writer
from langgraph_openai_serve import citation_event

get_stream_writer()(
    citation_event(
        url="https://example.com/source",
        title="Example source",
        span=(10, 14),
    )
)

span uses Python's half-open convention, so text[10:14] returns the cited text. LGOS converts it to OpenAI's inclusive end_index at the event boundary. Use citation_slice(annotation, text) to validate received indices and convert them back to a Python slice. Citation events must refer to the final rendered assistant text.

See Citation ownership for transport and client behavior.

The graph runner preserves LangGraph's native CustomStreamPart values, including their execution namespace. Other event types remain available to direct runner consumers through langgraph_openai_serve.graph.runner.

Demo Models¶

make run-demo-api registers:

• simple-graph (runtime settings for conversation history and intended audience)
• citation-events (structured URL citations alongside portable Markdown)
• lgos-rag
• custom-input-output-context
• advanced-mcp-tools
• complex-subgraphs
• custom-event-showcase (status, progress, and artifact stream events)
• interruptible-approval

Local Commands¶

uv sync --frozen
make help

make run-demo-api

make -s test
make test-bifrost
uv run --module pytest tests/path/to/test_file.py
uv run --module pytest tests/path/to/test_file.py::test_name
make -s lint
make -s format

make doc-build
make doc-serve

defaults()

29
30
31
32
33
34
35

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

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

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

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

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

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

build_context(request, graph)

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

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

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

    # LangGraph owns context_schema coercion when the graph is invoked.
    return context

build_input(request, messages)

108
109
110
111
112
113
114
115
116

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

render_output(output)

144
145
146
147
148
149

async def render_output(self, output: Any) -> str:
    """Convert native graph output into assistant response text."""
    if self.output_to_text is None:
        messages = output["messages"]
        return messages[-1].content if messages else ""
    return await _maybe_await(self.output_to_text(output))

resolve_graph()

 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

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

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

    if self.supports(GraphFeature.INTERRUPTS) and graph.checkpointer is None:
        raise GraphConfigurationError(
            "Interrupt-enabled graphs must be compiled with a checkpointer."
        )

    return graph

supports(feature)

80
81
82

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

validate_client_settings(value)

71
72
73
74
75
76
77
78

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

get_graph(name)

190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

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.
    """
    if name not in self.registry:
        raise GraphNotFoundError(f"Graph '{name}' not found in registry.")
    return self.registry[name]

get_graph_names()

186
187
188

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

register(model_id, config)

182
183
184

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

LanggraphOpenaiServe(graphs, app=None)

53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

def __init__(
    self,
    graphs: GraphRegistry,
    app: FastAPI | 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.

    Raises:
        TypeError: If graphs is not a GraphRegistry instance.
    """
    if not isinstance(graphs, GraphRegistry):
        raise TypeError(
            "Invalid type for graphs parameter. Expected GraphRegistry."
        )

    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

    logger.info("Using provided GraphRegistry instance")
    self.graph_registry = graphs

    # Attach the registry to the host app for callers that inspect app state.
    self.app.state.graph_registry = self.graph_registry

    logger.info(
        f"Initialized LanggraphOpenaiServe with {len(self.graph_registry.registry)} graphs"
    )
    logger.info(
        f"Available graphs: {', '.join(self.graph_registry.get_graph_names())}"
    )

bind_openai_api(prefix=None)

 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

def bind_openai_api(self, prefix: str | None = None):
    """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.
    """
    if prefix is None:
        prefix = settings.OPENAI_API_PREFIX
    else:
        prefix = normalize_openai_api_prefix(prefix)

    docs_url = "/docs" if settings.OPENAI_API_DOCS_ENABLED else None
    redoc_url = "/redoc" if settings.OPENAI_API_DOCS_ENABLED else None
    openapi_url = "/openapi.json" if settings.OPENAI_API_DOCS_ENABLED else None

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

    self.app.mount(prefix, openai_app, name="openai")

    logger.info(f"Bound OpenAI chat completion endpoints with prefix: {prefix}")

    return self

citation_event(*, url, title, span)

61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

def citation_event(
    *,
    url: str,
    title: str,
    span: tuple[int, int],
) -> dict[str, object]:
    """Build an OpenAI URL citation from a Python-style half-open span."""
    start, stop = span
    if not 0 <= start < stop:
        raise ValueError("citation span must define a valid non-empty text range")

    return Annotation(
        type="url_citation",
        url_citation=AnnotationURLCitation(
            url=url,
            title=title,
            start_index=start,
            end_index=stop - 1,
        ),
    ).model_dump(mode="json")

citation_slice(annotation, content)

83
84
85
86
87
88
89
90

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

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

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

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

reject_legacy_fields(data)

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

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

reject_legacy_fields(data)

 93
 94
 95
 96
 97
 98
 99
100
101

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

generate_completion(chat_request, run)

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

async def generate_completion(
    self, chat_request: ChatCompletionRequest, run: GraphRun
) -> ChatCompletionResponse:
    """Generate a chat completion."""
    start_time = time.time()

    invocation = await invoke_run(run)
    completion = invocation.output
    tokens_used = usage_for(completion, chat_request.messages)
    annotations = (
        [
            annotation
            for event in invocation.custom_events
            if (annotation := annotation_from_custom_event(event, completion))
            is not None
        ]
        if isinstance(completion, str)
        else []
    )

    response = chat_completion_response(
        model=chat_request.model,
        completion=completion,
        annotations=annotations,
        usage=tokens_used,
    )

    logger.info(
        f"Chat completion finished in {time.time() - start_time:.2f}s. "
        f"Total tokens: {tokens_used['total_tokens']}"
    )

    return response

stream_completion(chat_request, run)

 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
114
115
116
117
118
119
120
121
122

async def stream_completion(
    self, chat_request: ChatCompletionRequest, run: GraphRun
) -> AsyncGenerator[str, None]:
    """Stream a chat completion response."""
    start_time = time.time()
    response_builder = ChatCompletionStreamResponseBuilder(chat_request.model)
    custom_events: list[CustomStreamPart] = []
    content_parts: list[str] = []
    include_client_events = stream_events_requested(chat_request.metadata)

    try:
        yield response_builder.role()

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

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

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

        content = "".join(content_parts)
        annotations = [
            annotation
            for event in custom_events
            if (annotation := annotation_from_custom_event(event, content))
            is not None
        ]
        yield response_builder.finish("stop", annotations=annotations)
        yield response_builder.done()

        logger.info(
            f"Streamed chat completion finished in {time.time() - start_time:.2f}s"
        )

    except Exception as e:
        logger.exception("Error streaming chat completion")
        yield response_builder.error(str(e))
        yield response_builder.done()

annotation_from_custom_event(event, content)

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

def annotation_from_custom_event(
    event: CustomStreamPart,
    content: str,
) -> Annotation | None:
    """Validate a recognized OpenAI annotation custom event."""
    payload = event["data"]
    if not isinstance(payload, dict) or payload.get("type") != "url_citation":
        return None

    annotation = Annotation.model_validate(payload)
    citation_slice(annotation, content)
    return annotation

client_event_extension_from_custom_event(event)

22
23
24
25
26
27
28

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

stream_events_requested(metadata)

15
16
17
18
19

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

ChatCompletionStreamResponseBuilder(model)

107
108
109
110

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

client_event(extension)

118
119
120
121
122
123

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

chat_completion_response(*, model, completion, annotations=None, usage)

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

def chat_completion_response(
    *,
    model: str,
    completion: LangGraphOutput,
    annotations: list[Annotation] | None = None,
    usage: dict[str, int],
) -> ChatCompletionResponse:
    """Build a non-streaming OpenAI-compatible chat completion response."""
    message, finish_reason = response_message(completion, annotations)
    return ChatCompletionResponse(
        id=f"chatcmpl-{uuid.uuid4()}",
        created=int(time.time()),
        model=model,
        choices=[
            ChatCompletionResponseChoice(
                index=0,
                message=message,
                finish_reason=finish_reason,
            )
        ],
        usage=UsageInfo(
            prompt_tokens=usage["prompt_tokens"],
            completion_tokens=usage["completion_tokens"],
            total_tokens=usage["total_tokens"],
        ),
    )

create_chat_completion(chat_request, service, graph_registry, stream_owner)

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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

@router.post("/chat/completions", response_model=ChatCompletionResponse)
async def create_chat_completion(
    chat_request: ChatCompletionRequest,
    service: Annotated[ChatCompletionService, Depends(ChatCompletionService)],
    graph_registry: Annotated[GraphRegistry, Depends(get_graph_registry_dependency)],
    stream_owner: Annotated[
        _StreamOwner,
        Depends(_stream_owner_dependency, scope="request"),
    ],
) -> StreamingResponse | ChatCompletionResponse:
    """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.
        service: The chat completion service dependency.
        stream_owner: The request-scoped streaming task owner.

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

    logger.info(
        f"Received chat completion request for model: {chat_request.model}, "
        f"stream: {chat_request.stream}"
    )

    try:
        run = await prepare_run(
            chat_request.model,
            chat_request.messages,
            graph_registry,
            chat_request,
        )

        if chat_request.stream:
            logger.info("Streaming chat completion response")
            body = stream_owner.start(service.stream_completion(chat_request, run))
            return StreamingResponse(
                body,
                media_type="text/event-stream",
            )

        logger.info("Generating non-streaming chat completion response")
        response = await service.generate_completion(chat_request, run)
    except CLIENT_ERROR_TYPES as e:
        raise OpenAIHTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            error=ErrorObject(
                message=str(e),
                type="invalid_request_error",
                param=client_error_param(e),
            ),
        ) from e
    except GraphConfigurationError as e:
        raise OpenAIHTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            error=ErrorObject(
                message=str(e),
                type="server_error",
            ),
        ) from e
    logger.info("Returning non-streaming chat completion response")
    return response

health_check()

 8
 9
10
11
12
13

@router.get("/health")
def health_check() -> None:
    """Checks the health of a project.

    It returns 200 if the project is healthy.
    """

version()

16
17
18
19

@router.get("/version")
def version() -> dict[str, str]:
    """Return API version."""
    return {"version": get_version()}

get_model(model, graph_registry)

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

def get_model(self, model: str, graph_registry: GraphRegistry) -> ModelDetails:
    """Get one registered graph as an OpenAI model with LGOS metadata."""
    graph_config = graph_registry.get_graph(model)
    client_settings = graph_config.client_settings
    client_settings_details = None
    if client_settings is not None:
        client_settings_details = ModelClientSettings(
            json_schema=client_settings_json_schema(client_settings),
            defaults=client_settings_default_values(client_settings),
        )

    details = ModelDetails(
        id=model,
        created=MODEL_CREATED,
        owned_by=MODEL_OWNER,
        langgraph_openai_serve=LangGraphModelExtension(
            features=sorted(
                graph_config.features,
                key=lambda feature: feature.value,
            ),
            client_settings=client_settings_details,
        ),
    )
    logger.info(f"Retrieved model details for {model}")
    return details

get_models(graph_registry)

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

def get_models(self, graph_registry: GraphRegistry) -> ModelList:
    """Get a list of available models.

    Args:
        graph_registry: The GraphRegistry containing registered graphs.

    Returns:
        A list of models in OpenAI compatible format.
    """
    models = [
        Model(
            id=name,
            created=MODEL_CREATED,
            owned_by=MODEL_OWNER,
        )
        for name in graph_registry.registry
    ]

    logger.info(f"Retrieved {len(models)} available models")
    return ModelList(data=models)

get_graph_registry_dependency(request)

26
27
28

def get_graph_registry_dependency(request: Request) -> GraphRegistry:
    """Dependency to get the graph registry from the app state."""
    return request.app.state.graph_registry

list_models(service, graph_registry)

31
32
33
34
35
36
37
38
39
40

@router.get("", response_model=ModelList)
def list_models(
    service: Annotated[ModelService, Depends(ModelService)],
    graph_registry: Annotated[GraphRegistry, Depends(get_graph_registry_dependency)],
):
    """Get a list of available models."""
    logger.info("Received request to list models")
    models = service.get_models(graph_registry)
    logger.info(f"Returning {len(models.data)} models")
    return models

retrieve_model(model, service, graph_registry)

43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

@router.get(
    "/{model}",
    response_model=ModelDetails,
    response_model_exclude_none=True,
)
def retrieve_model(
    model: str,
    service: Annotated[ModelService, Depends(ModelService)],
    graph_registry: Annotated[GraphRegistry, Depends(get_graph_registry_dependency)],
) -> ModelDetails:
    """Retrieve one registered graph as an OpenAI model."""
    logger.info(f"Received request to retrieve model: {model}")
    try:
        return service.get_model(model, graph_registry)
    except GraphNotFoundError as exc:
        raise OpenAIHTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            error=ErrorObject(
                message=str(exc),
                type="invalid_request_error",
                param="model",
                code="model_not_found",
            ),
        ) from exc

OpenAIHTTPException(*, status_code, error, headers=None)

18
19
20
21
22
23
24
25
26
27
28
29
30

def __init__(
    self,
    *,
    status_code: int,
    error: ErrorObject,
    headers: dict[str, str] | None = None,
) -> None:
    super().__init__(
        status_code=status_code,
        detail=error.message,
        headers=headers,
    )
    self.error = error

configure_openai_error_handlers(app)

33
34
35
36
37
38
39
40
41

def configure_openai_error_handlers(app: FastAPI) -> None:
    """Install OpenAI-compatible JSON error handlers on a FastAPI app."""
    app.add_exception_handler(OpenAIHTTPException, openai_http_exception_handler)
    app.add_exception_handler(StarletteHTTPException, openai_http_exception_handler)
    app.add_exception_handler(
        RequestValidationError,
        openai_request_validation_exception_handler,
    )
    app.add_exception_handler(Exception, openai_unhandled_exception_handler)

check_langfuse_settings(v)

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

@field_validator("ENABLE_LANGFUSE")
@classmethod
def check_langfuse_settings(cls, v: bool) -> bool:
    """Validate Langfuse settings if enabled."""
    if v is False:
        return v

    # Check if langfuse package is installed
    if importlib.util.find_spec("langfuse") is None:
        raise RuntimeError(
            "Langfuse is enabled but the 'langfuse' package is not installed. "
            "Please install it, e.g., with `uv add langgraph-openai-serve[tracing]`."
        )

    # Check for required environment variables
    required_env_vars = [
        "LANGFUSE_BASE_URL",
        "LANGFUSE_PUBLIC_KEY",
        "LANGFUSE_SECRET_KEY",
    ]
    missing_vars = [var for var in required_env_vars if os.getenv(var) is None]

    if missing_vars:
        raise RuntimeError(
            "Langfuse is enabled but the following environment variables are not set: "
            f"{', '.join(missing_vars)}. Please set these variables."
        )

    return v

validate_openai_api_prefix(v)

44
45
46
47
48

@field_validator("OPENAI_API_PREFIX")
@classmethod
def validate_openai_api_prefix(cls, v: str) -> str:
    """Validate the mount prefix for OpenAI-compatible endpoints."""
    return normalize_openai_api_prefix(v)

normalize_openai_api_prefix(v)

 8
 9
10
11
12
13
14
15
16
17

def normalize_openai_api_prefix(v: str) -> str:
    """Normalize and validate the OpenAI-compatible API mount prefix."""
    if not v.startswith("/"):
        raise ValueError("OPENAI_API_PREFIX must start with '/'.")
    if len(v) > 1:
        normalized = v.rstrip("/")
        if not normalized:
            raise ValueError("OPENAI_API_PREFIX must not contain only slashes.")
        return normalized
    return v