API Reference

LangGraphApp

LangGraphApp(auth_level=func.AuthLevel.FUNCTION, health_auth_level=func.AuthLevel.ANONYMOUS, health_details_auth_level=None, max_stream_response_bytes=1024 * 1024, max_request_body_bytes=1024 * 1024, max_input_depth=32, max_input_nodes=10000, platform_compat=False, async_runs=False, thread_lock=None, observer=None, route_prefix=_ROUTE_PREFIX, _registrations=dict(), _sb_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 — anonymous liveness probe (returns only {"status": "ok"})
• GET /api/health/details — registered-graph inventory (protected; gated by health_details_auth_level)
• 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

All /stream responses are buffered SSE — chunks emitted by the graph are collected during execution and flushed as SSE events after the run completes, so clients do not receive partial tokens incrementally. This is a deliberate consequence of the classic HttpRequest/HttpResponse routing model this package is built on, not a platform limitation. Azure Functions Python v2 does support true HTTP streaming (runtime 4.34.1+) via the azurefunctions-extensions-http-fastapi ASGI extension, but enabling it switches the entire function app to the FastAPI/ASGI model, which cannot be mixed with the classic routes used here — an app-wide architectural change tracked in issue #378 (see the "Streaming: buffered SSE and the true-streaming migration" section of DESIGN.md).

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. The anonymous GET /api/health returns only {"status": "ok"} — it never enumerates registered graphs. The detailed inventory (graph names, descriptions, and checkpointer status) lives on a separate GET /api/health/details endpoint gated by health_details_auth_level, which defaults to the app-level auth_level (FUNCTION) so the inventory is protected unless explicitly opened up.

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, async_mode=False)

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). async_mode: When True, route this graph's invoke/stream endpoints through async Azure Functions handlers that await the graph's ainvoke / astream methods. Defaults to False (the sync path), so existing CompiledStateGraph deployments — which expose both sync and async methods — keep their current behavior. A graph that exposes only async methods is always served async, regardless of this flag.

    Raises:
        TypeError: If *graph* does not satisfy the required protocol, or if
            ``request_model``/``response_model`` is provided but is not a
            Pydantic ``BaseModel`` subclass.
        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,
        async_mode: bool = False,
    ) -> 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).
            async_mode: When ``True``, route this graph's invoke/stream endpoints
                through async Azure Functions handlers that ``await`` the graph's
                ``ainvoke`` / ``astream`` methods. Defaults to ``False`` (the sync
                path), so existing ``CompiledStateGraph`` deployments — which expose
                both sync and async methods — keep their current behavior. A graph
                that exposes only async methods is always served async, regardless
                of this flag.

        Raises:
            TypeError: If *graph* does not satisfy the required protocol, or if
                ``request_model``/``response_model`` is provided but is not a
                Pydantic ``BaseModel`` subclass.
            ValueError: If *name* is already registered or invalid.
        """
        has_sync_invoke = isinstance(graph, InvocableGraph)
        has_async_invoke = isinstance(graph, AsyncInvocableGraph)
        if async_mode and not has_async_invoke:
            raise TypeError(
                "async_mode=True requires an ainvoke() method. "
                f"Got {type(graph).__name__}"
            )
        if not has_sync_invoke and not has_async_invoke:
            raise TypeError(
                "Graph must have an invoke() or ainvoke() method. "
                f"Got {type(graph).__name__}"
            )
        if self.platform_compat and has_async_invoke and not has_sync_invoke:
            raise TypeError(
                f"Graph {name!r} exposes only async methods (ainvoke/astream), "
                "but platform_compat=True. LangGraph Platform-compatible runs "
                "(/runs/wait, /runs/stream, /threads/{id}/runs/*) execute graphs "
                "through synchronous invoke()/stream() calls, so an async-only "
                "graph cannot be served on the Platform surface. Provide a graph "
                "that also exposes invoke()/stream() (e.g. a standard compiled "
                "LangGraph graph), or register it on a LangGraphApp without "
                "platform_compat=True to use the native async endpoints."
            )
        _validate_optional_model(request_model, "request_model")
        _validate_optional_model(response_model, "response_model")
        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,
            async_mode=async_mode,
            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

Detailed health response (protected /health/details surface).

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
str | None	An opaque, non-empty owner token if the lock was acquired, or
str | None	None if it is held elsewhere (in the same process or, for
str | None	distributed backends, on another Function App instance). The token
str | None	is truthy on success and None on failure, so callers may still
str | None	branch on truthiness. Pass the returned token back to
str | None	meth:release so a stale caller cannot free a lock that has since
str | None	been re-acquired by a different execution.

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

def acquire(self, graph_name: str, thread_id: str, timeout: float = 0.0) -> str | None:
    """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:
        An opaque, non-empty **owner token** if the lock was acquired, or
        ``None`` if it is held elsewhere (in the same process or, for
        distributed backends, on another Function App instance). The token
        is truthy on success and ``None`` on failure, so callers may still
        branch on truthiness. Pass the returned token back to
        :meth:`release` so a stale caller cannot free a lock that has since
        been re-acquired by a different execution.
    """
    ...

release(graph_name, thread_id, token)

Release a lock previously acquired with :meth:acquire.

Parameters:

Name	Type	Description	Default
graph_name	str	Registered graph name.	required
thread_id	str	Thread ID that was locked.	required
token	str	The owner token returned by the matching :meth:acquire call. If it does not match the currently-held owner (e.g. the lock was dropped and re-acquired by a newer execution), release is a no-op logged at DEBUG — the newer owner is preserved.	required

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, token: str) -> None:
    """Release a lock previously acquired with :meth:`acquire`.

    Args:
        graph_name: Registered graph name.
        thread_id: Thread ID that was locked.
        token: The owner token returned by the matching :meth:`acquire`
            call. If it does not match the currently-held owner (e.g. the
            lock was dropped and re-acquired by a newer execution), release
            is a **no-op** logged at DEBUG — the newer owner is preserved.

    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._states: dict[tuple[str, str], _KeyState] = {}
    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. Returns a fresh opaque owner token on success, or None if the lock is held.

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

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

    See :meth:`ThreadLock.acquire` for the general contract. Returns a
    fresh opaque owner token on success, or ``None`` if the lock is held.
    """
    key = (graph_name, thread_id)
    with self._guard:
        state = self._states.get(key)
        if state is None:
            state = _KeyState()
            self._states[key] = state
        # Reserve a ref BEFORE dropping the guard so a concurrent release
        # cannot garbage-collect this state while we block on it below.
        state.refs += 1
    if timeout > 0.0:
        acquired = state.lock.acquire(blocking=True, timeout=timeout)
    else:
        acquired = state.lock.acquire(blocking=False)
    if not acquired:
        with self._guard:
            state.refs -= 1
            self._maybe_gc(key, state)
        return None
    token = secrets.token_hex(16)
    with self._guard:
        state.token = token
    return token

release(graph_name, thread_id, token)

Release the lock for (graph_name, thread_id).

See :meth:ThreadLock.release for the general contract. A token that does not match the currently-held owner is a no-op (DEBUG log), so a stale caller cannot release a lock re-acquired by a newer execution. On a successful match the owner ref is dropped and the underlying state is garbage-collected when no other request 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, token: str) -> None:
    """Release the lock for ``(graph_name, thread_id)``.

    See :meth:`ThreadLock.release` for the general contract. A ``token``
    that does not match the currently-held owner is a no-op (DEBUG log),
    so a stale caller cannot release a lock re-acquired by a newer
    execution. On a successful match the owner ref is dropped and the
    underlying state is garbage-collected when no other request holds it,
    so long-lived workers do not grow the internal dict unboundedly.
    """
    key = (graph_name, thread_id)
    with self._guard:
        state = self._states.get(key)
        current_token = state.token if state is not None else None
    if state is None:
        logger.debug(
            "release() called for unknown lock key %s/%s; ignoring", graph_name, thread_id
        )
        return
    if current_token != token:
        logger.debug(
            "release() owner mismatch for %s/%s; lock is held by a newer "
            "owner, ignoring stale release",
            graph_name,
            thread_id,
        )
        return
    try:
        state.lock.release()
    except RuntimeError:
        # Not held (or not held by us). Log and fall through so the ref is
        # still dropped and the entry is cleaned up — never leak the state.
        logger.debug(
            "release() called on unheld lock for %s/%s; cleaning up",
            graph_name,
            thread_id,
        )
    with self._guard:
        state.token = None
        state.refs -= 1
        self._maybe_gc(key, state)

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], _LeaseState] = {}
    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
    # Per-call client-side timeout for a single ``renew()`` and the pool used
    # to run renewals concurrently. Both stay unset when auto-renewal is off.
    self._renewal_call_timeout: float = 0.0
    self._renewal_executor: ThreadPoolExecutor | None = None

    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
        # Bound each renew() by one renewal interval: a call still pending
        # when the next tick is due is abandoned (counted as a transient
        # failure) so a hung call never stalls the loop or later ticks.
        self._renewal_call_timeout = self._renewal_interval
        self._renewal_executor = ThreadPoolExecutor(
            max_workers=_RENEWAL_MAX_WORKERS,
            thread_name_prefix=f"azblob-lease-renew-{id(self):x}",
        )
        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.

A key that is already tracked locally — including one whose lease has been marked lost but not yet released — returns None immediately without hitting Azure, so the original execution retains exclusive local ownership until it releases.

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

def acquire(self, graph_name: str, thread_id: str, timeout: float = 0.0) -> str | None:
    """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.

    A key that is already tracked locally — **including** one whose lease
    has been marked ``lost`` but not yet released — returns ``None``
    immediately without hitting Azure, so the original execution retains
    exclusive local ownership until it releases.
    """
    key = (graph_name, thread_id)
    # Fast local check — do not hammer Azure if we already track a lease
    # (a lost-but-unreleased entry still counts as occupied).
    with self._active_leases_guard:
        if key in self._active_leases:
            return None

    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 None
            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 None
            state = _LeaseState(lease=lease)
            self._active_leases[key] = state
        return state.token

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. Entries already marked lost are dropped without a service-side release (their lease is gone).

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. Entries already marked ``lost`` are dropped without a
    service-side release (their lease is gone).
    """
    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)
    executor = self._renewal_executor
    if executor is not None:
        # Don't wait on in-flight renewals — a hung renew() must not block
        # shutdown; cancel_futures drops any that never started running.
        executor.shutdown(wait=False, cancel_futures=True)
    with self._active_leases_guard:
        remaining = list(self._active_leases.items())
        self._active_leases.clear()
    for key, state in remaining:
        if state.lost:
            continue
        try:
            state.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, token)

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

Best-effort — never raises. If token does not match the owner token of the currently-tracked lease (e.g. the key was dropped and re-acquired by a newer execution), release is a no-op logged at DEBUG so the newer owner is preserved. If the tracked lease was already marked lost (definitive loss or exhausted renewals), the local entry is simply cleared without calling lease.release() because the lease is gone. Otherwise the lease is released best-effort; 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, token: str) -> None:
    """Release the Azure Blob lease for ``(graph_name, thread_id)``.

    Best-effort — never raises. If ``token`` does not match the owner
    token of the currently-tracked lease (e.g. the key was dropped and
    re-acquired by a newer execution), release is a no-op logged at DEBUG
    so the newer owner is preserved. If the tracked lease was already
    marked ``lost`` (definitive loss or exhausted renewals), the local
    entry is simply cleared without calling ``lease.release()`` because
    the lease is gone. Otherwise the lease is released best-effort;
    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:
        state = self._active_leases.get(key)
        if state is None:
            logger.debug(
                "release() called for unknown lease key %s/%s; ignoring",
                graph_name,
                thread_id,
            )
            return
        if state.token != token:
            logger.debug(
                "release() owner mismatch for %s/%s; lease is held by a "
                "newer owner, ignoring stale release",
                graph_name,
                thread_id,
            )
            return
        self._active_leases.pop(key, None)
    if state.lost:
        # Lease already gone — nothing to release on the service side.
        logger.debug(
            "release() clearing lost lease entry for %s/%s (no service release)",
            graph_name,
            thread_id,
        )
        return
    try:
        state.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