API Reference

LangGraphApp

LangGraphApp(auth_level=func.AuthLevel.FUNCTION, health_auth_level=func.AuthLevel.ANONYMOUS, max_stream_response_bytes=1024 * 1024, max_request_body_bytes=1024 * 1024, max_input_depth=32, max_input_nodes=10000, platform_compat=False, thread_lock=None, route_prefix=_ROUTE_PREFIX, _registrations=dict()) dataclass

Wraps LangGraph compiled graphs into Azure Functions HTTP endpoints.

Usage::

from azure_functions_langgraph import LangGraphApp

app = LangGraphApp()
app.register(graph=compiled_graph, name="my_agent")
func_app = app.function_app

This auto-registers:

• POST /api/graphs/{name}/invoke — synchronous invocation
• POST /api/graphs/{name}/stream — buffered SSE response (not true streaming)
• GET /api/health — health check with registered graph list
• GET /api/graphs/{name}/threads/{thread_id}/state — thread state (StatefulGraph only)

Note

The graph argument must satisfy the :class:LangGraphLike protocol (i.e. have .invoke() and .stream() methods). This avoids a hard import dependency on langgraph at the library level.

Note

v0.1 streams are buffered — all chunks are collected and returned in a single SSE-formatted HTTP response. True streaming (chunked transfer encoding) is planned for a future release once Azure Functions Python HTTP streaming stabilises.

Note

The default auth_level is :attr:~azure.functions.AuthLevel.FUNCTION, so deployed endpoints require a function key by default. Pass auth_level=func.AuthLevel.ANONYMOUS explicitly for public access (e.g. local development); doing so emits an unconditional UserWarning. The health_auth_level parameter controls the auth level of the health endpoint independently and defaults to :attr:~azure.functions.AuthLevel.ANONYMOUS, which is the conventional choice for liveness/readiness probes.

Note

Per-thread locking on the native invoke/stream endpoints is pluggable via :attr:thread_lock. The default, :class:~azure_functions_langgraph.locks.inprocess.InProcessThreadLock, is in-process only — it is not distributed across Function App instances, worker processes, or hosts. For multi-instance production deployments, supply a distributed backend such as :class:~azure_functions_langgraph.locks.azure_blob.AzureBlobLeaseThreadLock, or (for platform-compat runs) enable platform_compat=True with :class:~azure_functions_langgraph.stores.azure_table.AzureTableThreadStore, which provides ETag-based atomic locking. Set the AZFUNC_LANGGRAPH_LOCK_BACKEND environment variable to distributed to fail-fast at construction if the default in-process backend is still wired.

function_app property

Return an azure.functions.FunctionApp with all routes registered.

register(graph, name, description=None, stream=True, auth_level=None, *, request_model=None, response_model=None)

Register a compiled LangGraph graph.

Parameters:

Name	Type	Description	Default
graph	Any	Any object satisfying :class:~protocols.LangGraphLike (typically a CompiledStateGraph from langgraph).	required
name	str	Unique name for this graph (used in URL routes).	required
description	Optional[str]	Optional human-readable description.	None
stream	bool	Whether to enable the stream endpoint for this graph.	True
auth_level	Optional[AuthLevel]	Override app-level auth for this graph's endpoints. When None (default), the app-level auth_level is used.	None
request_model	Optional[type[Any]]	Optional Pydantic model class for request body (used by the metadata / bridge API, not for runtime validation).	None
response_model	Optional[type[Any]]	Optional Pydantic model class for response body (used by the metadata / bridge API, not for runtime validation).	None

Raises:

Type	Description
TypeError	If graph does not satisfy the required protocol.
ValueError	If name is already registered or invalid.

Source code in src/azure_functions_langgraph/app.py

def register(
    self,
    graph: Any,
    name: str,
    description: Optional[str] = None,
    stream: bool = True,
    auth_level: Optional[func.AuthLevel] = None,
    *,
    request_model: Optional[type[Any]] = None,
    response_model: Optional[type[Any]] = None,
) -> None:
    """Register a compiled LangGraph graph.

    Args:
        graph: Any object satisfying :class:`~protocols.LangGraphLike`
            (typically a ``CompiledStateGraph`` from ``langgraph``).
        name: Unique name for this graph (used in URL routes).
        description: Optional human-readable description.
        stream: Whether to enable the stream endpoint for this graph.
        auth_level: Override app-level auth for this graph's endpoints.
            When ``None`` (default), the app-level ``auth_level`` is used.
        request_model: Optional Pydantic model class for request body
            (used by the metadata / bridge API, not for runtime validation).
        response_model: Optional Pydantic model class for response body
            (used by the metadata / bridge API, not for runtime validation).

    Raises:
        TypeError: If *graph* does not satisfy the required protocol.
        ValueError: If *name* is already registered or invalid.
    """
    if not isinstance(graph, InvocableGraph):
        raise TypeError(f"Graph must have an invoke() method. Got {type(graph).__name__}")
    name_err = validate_graph_name(name)
    if name_err:
        raise ValueError(name_err)
    if name in self._registrations:
        raise ValueError(f"Graph {name!r} is already registered")
    self._registrations[name] = _GraphRegistration(
        graph=graph,
        name=name,
        description=description,
        stream_enabled=stream,
        auth_level=auth_level,
        request_model=request_model,
        response_model=response_model,
    )
    # Reset cached function app so routes are re-generated
    self._function_app = None

Request models

InvokeRequest

InvokeRequest

Bases: BaseModel

Request body for graph invocation.

StreamRequest

StreamRequest

Bases: BaseModel

Request body for graph streaming.

Response models

InvokeResponse

InvokeResponse

Bases: BaseModel

Response body for graph invocation.

HealthResponse

HealthResponse

Bases: BaseModel

Health check response.

ErrorResponse

ErrorResponse

Bases: BaseModel

Error response body.

GraphInfo

GraphInfo

Bases: BaseModel

Information about a registered graph.

Protocol interfaces

InvocableGraph

InvocableGraph

Bases: Protocol

Protocol for a graph that supports synchronous invocation.

StreamableGraph

StreamableGraph

Bases: Protocol

Protocol for a graph that supports synchronous streaming.

LangGraphLike

LangGraphLike

Bases: InvocableGraph, StreamableGraph, Protocol

Protocol combining invoke and stream — matches LangGraph's CompiledStateGraph.

Thread locks

Pluggable per-thread lock backends for the native invoke / stream endpoints. See Operations & API Surface for the operational guide (RBAC, lease renewal, production checklist).

ThreadLock

ThreadLock

Bases: Protocol

Contract for pluggable per-thread lock backends.

Implementations coordinate concurrent access to a native invoke/stream request that targets a specific (graph_name, thread_id) so that single-writer checkpointers (for example :class:~azure_functions_langgraph.checkpointers.azure_blob.AzureBlobCheckpointSaver) never see racing writes for one thread.

The two shipped implementations are :class:~azure_functions_langgraph.locks.inprocess.InProcessThreadLock (the default; single-process only) and :class:~azure_functions_langgraph.locks.azure_blob.AzureBlobLeaseThreadLock (distributed via Azure Blob lease CAS).

Third-party backends satisfying this protocol (Redis, Cosmos DB, etc.) can be plugged in via :attr:LangGraphApp.thread_lock.

acquire(graph_name, thread_id, timeout=0.0)

Attempt to acquire an exclusive lock for (graph_name, thread_id).

Parameters:

Name	Type	Description	Default
graph_name	str	Registered graph name.	required
thread_id	str	Thread ID drawn from config.configurable.thread_id.	required
timeout	float	Maximum seconds to wait for the lock. 0.0 (default) is non-blocking — matches the pre-existing native-endpoint behavior. Positive values block up to timeout seconds.	0.0

Returns:

Type	Description
bool	True if the lock was acquired, False if it is held
bool	elsewhere (in the same process or, for distributed backends, on
bool	another Function App instance).

Source code in src/azure_functions_langgraph/locks/base.py

def acquire(self, graph_name: str, thread_id: str, timeout: float = 0.0) -> bool:
    """Attempt to acquire an exclusive lock for ``(graph_name, thread_id)``.

    Args:
        graph_name: Registered graph name.
        thread_id: Thread ID drawn from ``config.configurable.thread_id``.
        timeout: Maximum seconds to wait for the lock. ``0.0`` (default)
            is non-blocking — matches the pre-existing native-endpoint
            behavior. Positive values block up to ``timeout`` seconds.

    Returns:
        ``True`` if the lock was acquired, ``False`` if it is held
        elsewhere (in the same process or, for distributed backends, on
        another Function App instance).
    """
    ...

release(graph_name, thread_id)

Release a previously acquired lock.

Must be safe to call even if the lock is not currently held by the caller — implementations should log at DEBUG level for any inconsistency rather than raising, so that the handler finally block never masks the underlying request failure.

Source code in src/azure_functions_langgraph/locks/base.py

def release(self, graph_name: str, thread_id: str) -> None:
    """Release a previously acquired lock.

    Must be safe to call even if the lock is not currently held by the
    caller — implementations should log at DEBUG level for any
    inconsistency rather than raising, so that the handler ``finally``
    block never masks the underlying request failure.
    """
    ...

InProcessThreadLock

InProcessThreadLock()

:class:threading.Lock-based per-thread lock scoped to a single worker.

This is the default backend when :attr:LangGraphApp.thread_lock is not supplied. It is not distributed — locks are held in this Python interpreter only and do not coordinate across:

• Multiple Azure Functions App instances (scale-out)
• Multiple worker processes on the same instance
• Warm swaps / cold starts

Multi-instance production deployments must supply a distributed backend such as :class:~azure_functions_langgraph.locks.azure_blob.AzureBlobLeaseThreadLock.

Thread-safety

Safe for concurrent acquire / release calls from multiple threads. Internal state is guarded by a private :class:threading.Lock.

Source code in src/azure_functions_langgraph/locks/inprocess.py

def __init__(self) -> None:
    self._locks: dict[tuple[str, str], threading.Lock] = {}
    self._guard = threading.Lock()

acquire(graph_name, thread_id, timeout=0.0)

Acquire the lock for (graph_name, thread_id).

See :meth:ThreadLock.acquire for the general contract.

Source code in src/azure_functions_langgraph/locks/inprocess.py

def acquire(self, graph_name: str, thread_id: str, timeout: float = 0.0) -> bool:
    """Acquire the lock for ``(graph_name, thread_id)``.

    See :meth:`ThreadLock.acquire` for the general contract.
    """
    with self._guard:
        lock = self._locks.setdefault((graph_name, thread_id), threading.Lock())
    if timeout > 0.0:
        return lock.acquire(blocking=True, timeout=timeout)
    return lock.acquire(blocking=False)

release(graph_name, thread_id)

Release the lock for (graph_name, thread_id).

See :meth:ThreadLock.release for the general contract. Also garbage-collects the underlying :class:threading.Lock when no other request currently holds it, so long-lived workers do not grow the internal dict unboundedly.

Source code in src/azure_functions_langgraph/locks/inprocess.py

def release(self, graph_name: str, thread_id: str) -> None:
    """Release the lock for ``(graph_name, thread_id)``.

    See :meth:`ThreadLock.release` for the general contract. Also
    garbage-collects the underlying :class:`threading.Lock` when no
    other request currently holds it, so long-lived workers do not
    grow the internal dict unboundedly.
    """
    key = (graph_name, thread_id)
    with self._guard:
        lock = self._locks.get(key)
    if lock is None:
        logger.debug(
            "release() called for unknown lock key %s/%s; ignoring", graph_name, thread_id
        )
        return
    try:
        lock.release()
    except RuntimeError:
        # Not held (or not held by us). Log and continue so the caller's
        # `finally` block never masks the original exception.
        logger.debug(
            "release() called on unheld lock for %s/%s; ignoring",
            graph_name,
            thread_id,
        )
        return
    # Clean up to prevent unbounded growth in long-lived workers.
    # Re-check under guard: only remove if the lock is not currently held
    # (another request may have acquired it between release and this check).
    with self._guard:
        current = self._locks.get(key)
        if current is lock and not lock.locked():
            self._locks.pop(key, None)

AzureBlobLeaseThreadLock

AzureBlobLeaseThreadLock(*, container_client, lease_duration=_LEASE_DURATION_MAX, blob_prefix='thread-locks/', auto_renew=True)

Distributed per-thread lock backed by Azure Blob leases.

Coordinates (graph_name, thread_id) locking across multiple Azure Functions instances by holding an exclusive lease on a marker blob per thread. Any :class:~azure.storage.blob.ContainerClient will do — the same container as the AzureBlobCheckpointSaver is a natural fit but a dedicated container is fine too.

.. warning:: By default this class renews Azure Blob leases in the background (a per-instance daemon thread renews every active lease at lease_duration / 3 intervals). Pass auto_renew=False to opt out — construction then emits a :class:UserWarning because a finite lease_duration can silently expire mid-execution and let another instance acquire the same (graph_name, thread_id) lock, allowing concurrent writes to single-writer checkpointers. Call :meth:close for graceful shutdown of the renewal thread when the lock instance is no longer needed; Azure Functions workers do not typically need this because the daemon thread dies when the interpreter exits.

Example

from azure.storage.blob import ContainerClient from azure_functions_langgraph import LangGraphApp from azure_functions_langgraph.locks import AzureBlobLeaseThreadLock

container = ContainerClient.from_connection_string(conn, "thread-locks") if not container.exists(): ... container.create_container() lock = AzureBlobLeaseThreadLock(container_client=container) app = LangGraphApp(thread_lock=lock)

Parameters:

Name	Type	Description	Default
container_client	_ContainerClientProtocol	An azure.storage.blob.ContainerClient bound to the container where marker blobs will live. The container must already exist — this class never creates it (that decision belongs to app-level infrastructure code).	required
lease_duration	int	Lease length in seconds. Must be 15-60 (finite) or -1 (infinite). Defaults to 60. With auto_renew=True (the default), a background daemon thread renews every active lease at lease_duration / 3 intervals, so execution time is no longer bounded by the lease. Set auto_renew=False to disable renewal — a finite lease then silently expires mid-execution and lets another instance acquire the same lock. Finite leases also auto-expire on the service if :meth:release never runs (host crash, scale-in), giving you a crash-recovery mechanism. Infinite leases require an operator to break them manually when a host crashes.	_LEASE_DURATION_MAX
blob_prefix	str	Prefix applied to every marker blob so lock blobs are visually grouped inside the container. Defaults to "thread-locks/".	'thread-locks/'
auto_renew	bool	If True (default), start a per-instance daemon thread that renews every active lease at lease_duration / 3 intervals until :meth:close (or process exit). If False, no renewal happens and construction emits a :class:UserWarning when lease_duration is finite, since finite leases will silently expire mid-execution. Ignored for lease_duration=-1 (infinite leases are not renewable).	True

Thread-safety

Safe for concurrent acquire / release calls from multiple threads. Only one thread in this process can hold a given lease at a time (the Azure API enforces this globally); this class enforces it locally by returning False from :meth:acquire when a lease is already tracked for the key.

Source code in src/azure_functions_langgraph/locks/azure_blob.py

def __init__(
    self,
    *,
    container_client: _ContainerClientProtocol,
    lease_duration: int = _LEASE_DURATION_MAX,
    blob_prefix: str = "thread-locks/",
    auto_renew: bool = True,
) -> None:
    if lease_duration != _LEASE_DURATION_INFINITE and not (
        _LEASE_DURATION_MIN <= lease_duration <= _LEASE_DURATION_MAX
    ):
        raise ValueError(
            f"lease_duration must be -1 (infinite) or between "
            f"{_LEASE_DURATION_MIN} and {_LEASE_DURATION_MAX} seconds; got {lease_duration}"
        )

    try:
        azure_blob_module = importlib.import_module("azure.storage.blob")
    except ImportError as exc:
        raise ImportError(
            "AzureBlobLeaseThreadLock requires optional dependency "
            "'azure-storage-blob'. Install with: "
            "pip install azure-functions-langgraph[azure-blob]"
        ) from exc

    azure_container_client = getattr(azure_blob_module, "ContainerClient", None)
    if azure_container_client is None or not isinstance(
        container_client, azure_container_client
    ):
        raise TypeError(
            "container_client must be an instance of azure.storage.blob.ContainerClient"
        )

    try:
        azure_core_exceptions = importlib.import_module("azure.core.exceptions")
    except (
        ImportError
    ) as exc:  # pragma: no cover - defensive; installed with azure-storage-blob
        raise ImportError(
            "AzureBlobLeaseThreadLock requires 'azure-core'. "
            "Install with: pip install azure-functions-langgraph[azure-blob]"
        ) from exc
    resource_exists_error = getattr(azure_core_exceptions, "ResourceExistsError", None)
    http_response_error = getattr(azure_core_exceptions, "HttpResponseError", None)
    if resource_exists_error is None or http_response_error is None:
        raise ImportError(  # pragma: no cover - defensive
            "azure.core.exceptions is missing ResourceExistsError or HttpResponseError; "
            "azure-core installation may be corrupt."
        )

    self._container_client: _ContainerClientProtocol = cast(
        _ContainerClientProtocol, container_client
    )
    self._lease_duration = lease_duration
    self._prefix = blob_prefix
    self._resource_exists_error: type[BaseException] = cast(
        type[BaseException], resource_exists_error
    )
    self._http_response_error: type[BaseException] = cast(
        type[BaseException], http_response_error
    )
    self._active_leases: dict[tuple[str, str], _BlobLeaseClientProtocol] = {}
    self._active_leases_guard = threading.Lock()

    self._auto_renew: bool = auto_renew and lease_duration != _LEASE_DURATION_INFINITE
    self._closed: bool = False
    self._shutdown_event: threading.Event = threading.Event()
    self._renewal_thread: threading.Thread | None = None
    self._renewal_interval: float = 0.0

    if lease_duration != _LEASE_DURATION_INFINITE and not auto_renew:
        warnings.warn(
            f"AzureBlobLeaseThreadLock(lease_duration={lease_duration}, "
            "auto_renew=False) is finite and auto-renewal is disabled. "
            "If a graph execution exceeds lease_duration seconds, the "
            "lease will silently expire mid-execution and another "
            "instance may acquire the same (graph_name, thread_id) "
            "lock, allowing concurrent writes to single-writer "
            "checkpointers. Enable auto_renew=True (the default) or "
            "pass lease_duration=-1 (infinite) whenever graph "
            "execution can exceed 60 seconds.",
            UserWarning,
            stacklevel=2,
        )

    if self._auto_renew:
        self._renewal_interval = lease_duration / _LEASE_RENEWAL_FRACTION
        self._renewal_thread = threading.Thread(
            target=self._renewal_worker,
            name=f"azblob-lease-renew-{id(self):x}",
            daemon=True,
        )
        self._renewal_thread.start()

acquire(graph_name, thread_id, timeout=0.0)

Attempt to hold an Azure Blob lease for (graph_name, thread_id).

Semantics match :meth:ThreadLock.acquire:

• timeout=0.0 — non-blocking. Returns immediately.
• timeout>0.0 — polls the Azure API with jittered backoff until the lease is acquired or the deadline expires.

Source code in src/azure_functions_langgraph/locks/azure_blob.py

def acquire(self, graph_name: str, thread_id: str, timeout: float = 0.0) -> bool:
    """Attempt to hold an Azure Blob lease for ``(graph_name, thread_id)``.

    Semantics match :meth:`ThreadLock.acquire`:

    * ``timeout=0.0`` — non-blocking. Returns immediately.
    * ``timeout>0.0`` — polls the Azure API with jittered backoff until
      the lease is acquired or the deadline expires.
    """
    key = (graph_name, thread_id)
    # Fast local check — do not hammer Azure if we already track a lease.
    with self._active_leases_guard:
        if key in self._active_leases:
            return False

    blob_client = self._container_client.get_blob_client(self._blob_name(graph_name, thread_id))
    self._ensure_marker(blob_client)

    deadline = time.monotonic() + timeout if timeout > 0.0 else 0.0
    while True:
        try:
            lease = blob_client.acquire_lease(lease_duration=self._lease_duration)
        except self._http_response_error as exc:
            if not self._is_lease_conflict(exc):
                raise
            if timeout <= 0.0 or time.monotonic() >= deadline:
                return False
            remaining = deadline - time.monotonic()
            time.sleep(min(_POLL_INTERVAL_MAX, max(_POLL_INTERVAL_MIN, remaining / 2)))
            continue

        with self._active_leases_guard:
            # Concurrent local acquire may have won the race — release the
            # lease we just took and report failure so callers stay
            # consistent with the fast-path check above.
            if key in self._active_leases:
                try:
                    lease.release()
                except Exception:  # pragma: no cover - defensive
                    logger.debug(
                        "Failed to release race-loser lease for %s/%s",
                        graph_name,
                        thread_id,
                        exc_info=True,
                    )
                return False
            self._active_leases[key] = lease
        return True

close()

Stop the renewal thread and release every active lease.

Idempotent and safe to call from any thread. After :meth:close, further :meth:acquire calls still work but will not be auto-renewed even if auto_renew=True was passed to the constructor.

Source code in src/azure_functions_langgraph/locks/azure_blob.py

def close(self) -> None:
    """Stop the renewal thread and release every active lease.

    Idempotent and safe to call from any thread. After :meth:`close`,
    further :meth:`acquire` calls still work but will not be
    auto-renewed even if ``auto_renew=True`` was passed to the
    constructor.
    """
    if self._closed:
        return
    self._closed = True
    self._shutdown_event.set()
    thread = self._renewal_thread
    if thread is not None and thread.is_alive():
        thread.join(timeout=_RENEWAL_SHUTDOWN_TIMEOUT)
    with self._active_leases_guard:
        remaining = list(self._active_leases.items())
        self._active_leases.clear()
    for key, lease in remaining:
        try:
            lease.release()
        except Exception:
            logger.debug(
                "Failed to release blob lease for %s/%s during close",
                key[0],
                key[1],
                exc_info=True,
            )

release(graph_name, thread_id)

Release the Azure Blob lease for (graph_name, thread_id).

Best-effort — never raises. Failures during release are logged at DEBUG and left for lease expiry (or manual break) to recover.

Source code in src/azure_functions_langgraph/locks/azure_blob.py

def release(self, graph_name: str, thread_id: str) -> None:
    """Release the Azure Blob lease for ``(graph_name, thread_id)``.

    Best-effort — never raises. Failures during release are logged at
    DEBUG and left for lease expiry (or manual break) to recover.
    """
    key = (graph_name, thread_id)
    with self._active_leases_guard:
        lease = self._active_leases.pop(key, None)
    if lease is None:
        logger.debug(
            "release() called for unknown lease key %s/%s; ignoring", graph_name, thread_id
        )
        return
    try:
        lease.release()
    except Exception:
        logger.debug(
            "Failed to release blob lease for %s/%s; will expire naturally",
            graph_name,
            thread_id,
            exc_info=True,
        )

OpenAPI integration

Bridges a LangGraphApp into azure-functions-openapi-python.

register_with_openapi

Registers all graph and app-level routes with the OpenAPI package and returns the number of routes registered. Requires the optional dependency azure-functions-openapi-python.

from azure_functions_langgraph import LangGraphApp
from azure_functions_langgraph.openapi import register_with_openapi

app = LangGraphApp()
# ... app.register(...) your graphs ...
count = register_with_openapi(app)  # -> int (routes registered)

register_with_openapi(app)

Register all graph and app-level endpoints with azure-functions-openapi-python.

Reads the metadata exposed by :meth:LangGraphApp.get_app_metadata and calls :func:azure_functions_openapi.register_openapi_metadata for each route.

.. deprecated:: 0.6.0 Request/response shape now also flows through the shared _azure_functions_metadata["endpoint"] namespace, which azure-functions-openapi reads directly from each handler (see issue #294; umbrella azure-functions-validation#270). This bridge is retained through the deprecation cycle to keep supplying documentation metadata (summary/description/tags) that the endpoint namespace intentionally does not carry.

Parameters:

Name	Type	Description	Default
app	LangGraphApp	A :class:LangGraphApp instance with graphs already registered.	required

Returns:

Type	Description
int	Number of routes registered with the openapi package.

Raises:

Type	Description
ImportError	If azure-functions-openapi-python is not installed.
TypeError	If a request_model or response_model is not a Pydantic BaseModel subclass.

Source code in src/azure_functions_langgraph/openapi.py

def register_with_openapi(app: LangGraphApp) -> int:
    """Register all graph and app-level endpoints with azure-functions-openapi-python.

    Reads the metadata exposed by :meth:`LangGraphApp.get_app_metadata` and
    calls :func:`azure_functions_openapi.register_openapi_metadata` for each
    route.

    .. deprecated:: 0.6.0
        Request/response *shape* now also flows through the shared
        ``_azure_functions_metadata["endpoint"]`` namespace, which
        ``azure-functions-openapi`` reads directly from each handler (see
        issue #294; umbrella azure-functions-validation#270). This bridge is
        retained through the deprecation cycle to keep supplying documentation
        metadata (``summary``/``description``/``tags``) that the endpoint
        namespace intentionally does not carry.

    Args:
        app: A :class:`LangGraphApp` instance with graphs already registered.

    Returns:
        Number of routes registered with the openapi package.

    Raises:
        ImportError: If ``azure-functions-openapi-python`` is not installed.
        TypeError: If a ``request_model`` or ``response_model`` is not a
            Pydantic ``BaseModel`` subclass.
    """
    try:
        from azure_functions_openapi import register_openapi_metadata
    except ImportError as exc:
        raise ImportError(
            "azure-functions-openapi-python is required for OpenAPI integration. "
            "Install it with: pip install azure-functions-openapi-python"
        ) from exc

    metadata = app.get_app_metadata()
    count = 0

    # Per-graph routes
    for graph_meta in metadata.graphs.values():
        for route in graph_meta.routes:
            request_body = None
            if route.request_model is not None:
                request_body = _build_request_body(route.request_model)

            response_model = route.response_model
            if response_model is not None:
                _validate_model(response_model, "response_model")
            register_openapi_metadata(
                path=route.path,
                method=route.method,
                summary=route.summary,
                description=route.description or graph_meta.description or "",
                tags=[graph_meta.name],
                request_body=request_body,
                response_model=response_model,
                parameters=list(route.parameters) if route.parameters else None,
            )
            count += 1

    # App-level routes (e.g. /health)
    for route in metadata.app_routes:
        register_openapi_metadata(
            path=route.path,
            method=route.method,
            summary=route.summary,
            description=route.description,
            tags=["system"],
            parameters=list(route.parameters) if route.parameters else None,
        )
        count += 1

    return count