Azure Functions LangGraph

⚠️ Experimental — pattern exploration. APIs and behavior may change. Not recommended as a production dependency yet.

Part of the Azure Functions Python DX Toolkit — dogfood-tested by azure-functions-cookbook-python.

Read this in: 한국어 | 日本語 | 简体中文

Stability — rolling 0.x, per surface. This package follows a rolling 0.x development model rather than a single "everything breaks until v1.0" contract. Core adapter APIs aim for backward compatibility across minor releases, while newer surfaces may still evolve. The per-surface stability table below is the single source of truth for what you can rely on; the Development Status :: 3 - Alpha classifier in pyproject.toml reflects the youngest surfaces, not the stable core. Please report issues on GitHub.

Surface	Tier	What it means
Core — LangGraphApp construction, native invoke / stream / state HTTP endpoints, auth levels	Stable	API frozen; a breaking change warrants a major bump
Checkpoint backends, thread locking, native async runtime (ainvoke/astream), version="v2" pass-through, RunObserver contract	Beta	Shape settled; may be refined before v1.0
LangGraph Platform compatibility, true HTTP streaming, Azure Service Bus trigger, Durable async run lifecycle	Experimental	May change or be removed between minor releases

Deploy LangGraph graphs as Azure Functions HTTP endpoints with minimal boilerplate.

Part of the Azure Functions Python DX Toolkit

Why this exists

Deploying LangGraph on Azure Functions is harder than it should be.

• LangGraph does not provide an Azure Functions-native deployment adapter
• Exposing compiled graphs as HTTP endpoints requires repetitive wiring
• Teams often rebuild the same invoke/stream wrapper for every project

This package provides a focused adapter for serving LangGraph graphs on Azure Functions Python v2.

What it does

• Zero-boilerplate deployment — register a compiled graph, get HTTP endpoints automatically
• Invoke endpoint — POST /api/graphs/{name}/invoke for synchronous execution
• Stream endpoint — POST /api/graphs/{name}/stream for buffered SSE responses (opt into true incremental streaming with StreamingLangGraphApp)
• Health endpoints — anonymous GET /api/health liveness probe ({"status": "ok"}, no graph inventory) plus protected GET /api/health/details listing registered graphs with checkpointer status
• Checkpointer pass-through — thread-based conversation state works via LangGraph's native config
• State endpoint — GET /api/graphs/{name}/threads/{thread_id}/state for thread state inspection (when supported)
• Per-graph auth — override app-level auth with register(..., auth_level=...)
• LangGraph Platform API compatibility — SDK-compatible endpoints for threads, runs, assistants, and state (v0.3+)
• Persistent storage backends — Azure Blob Storage checkpointer and Azure Table Storage thread store (v0.4+)

LangGraph Platform comparison

See COMPATIBILITY.md for the per-feature SDK support matrix, including which RunCreate fields, thread filters, and SDK calls return 501 Not Implemented.

Scope

• Azure Functions Python v2 programming model
• LangGraph graph deployment and HTTP exposure
• LangGraph runtime concerns: invoke, stream, threads, runs, and state
• Optional integration points for validation and OpenAPI via companion packages

This package is a deployment adapter — it wraps LangGraph, it does not replace it.

Internally, graph registration remains protocol-based (LangGraphLike), so any object satisfying the protocol works — but the package's documentation and examples focus on LangGraph use cases.

What this package does not do

This package does not own:

• OpenAPI document generation or Swagger UI — use azure-functions-openapi-python
• Request/response validation beyond LangGraph contracts — use azure-functions-validation-python
• Generic graph-serving abstractions beyond LangGraph

Note: For OpenAPI spec generation, use the azure-functions-openapi-python package with the bridge module (azure_functions_langgraph.openapi.register_with_openapi).

Installation

pip install azure-functions-langgraph

For persistent storage with Azure services:

# Azure Blob Storage checkpointer
pip install azure-functions-langgraph[azure-blob]

# Azure Table Storage thread store
pip install azure-functions-langgraph[azure-table]

# Both
pip install azure-functions-langgraph[azure-blob,azure-table]

For database or Cosmos DB checkpointer backends:

# Postgres checkpointer
pip install azure-functions-langgraph[postgres]

# SQLite checkpointer (local dev)
pip install azure-functions-langgraph[sqlite]

# Cosmos DB checkpointer
pip install azure-functions-langgraph[cosmos]

Your Azure Functions app should also include:

azure-functions
langgraph
azure-functions-langgraph

For local development:

git clone https://github.com/yeongseon/azure-functions-langgraph-python.git
cd azure-functions-langgraph
pip install -e .[dev]

Quick Start

New to this package? Follow the Deploy a LangGraph agent to Azure Functions in 5 minutes tutorial — a genuinely runnable, CI-smoke-tested walkthrough from a compiled graph to live HTTP endpoints, locally and on Azure.

Recommended learning path

Once the Quick Start echo agent runs, follow these four examples in order. Each one is the minimal delta over the previous step, so you add exactly one production concern at a time — real LLM → memory → tools → durable persistence:

This azure_openai_agent → conversation_memory → tool_calling_agent → production_persistent_agent progression takes you from a real agent to a durable, production-ready one end to end. Every example is a standalone Azure Functions app with a deterministic fake-model fallback, so all four pass CI without any cloud credentials.

from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

import azure.functions as func

from azure_functions_langgraph import LangGraphApp


# 1. Define your state
class AgentState(TypedDict):
    messages: list[dict[str, str]]


# 2. Define your nodes
def chat(state: AgentState) -> dict:
    user_msg = state["messages"][-1]["content"]
    return {"messages": state["messages"] + [{"role": "assistant", "content": f"Echo: {user_msg}"}]}


# 3. Build graph
builder = StateGraph(AgentState)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile()

# 4. Deploy (default is `AuthLevel.FUNCTION` — requires a function key when deployed).
app = LangGraphApp()
app.register(graph=graph, name="echo_agent")
func_app = app.function_app  # ← use this as your Azure Functions app

Start the Functions host locally:

func start

Verify locally and on Azure

After deploying (see docs/deployment.md), the same request produces the same response in both environments. Azure requires a function key (?code=<FUNCTION_KEY>) when auth_level is set to FUNCTION.

Local

curl -s http://localhost:7071/api/health

{"status": "ok"}

The anonymous liveness probe returns only {"status": "ok"}. To see the registered-graph inventory, call the protected details endpoint:

curl -s http://localhost:7071/api/health/details

{"status": "ok", "graphs": [{"name": "echo_agent", "description": null, "has_checkpointer": false}]}

Azure

# Liveness probe (anonymous by default)
curl -s "https://<your-app>.azurewebsites.net/api/health"

{"status": "ok"}

# Detailed inventory (protected — defaults to the app auth_level)
curl -s "https://<your-app>.azurewebsites.net/api/health/details?code=<FUNCTION_KEY>"

{"status": "ok", "graphs": [{"name": "echo_agent", "description": null, "has_checkpointer": false}]}

Response format verified against a temporary Azure Functions deployment of the simple_agent example in koreacentral (Python 3.12, Consumption plan). The Quick Start uses echo_agent for illustration; the health endpoint returns the same JSON structure regardless of graph name. URL anonymized.

Production authentication

Important: LangGraphApp defaults to AuthLevel.FUNCTION, so deployed endpoints require a function key (?code=<FUNCTION_KEY> or the x-functions-key header) out of the box. For an unauthenticated public surface — e.g. local development against func start when you also want to hit the endpoint without a key — pass auth_level=func.AuthLevel.ANONYMOUS explicitly. Doing so emits an unconditional UserWarning so an accidental anonymous deployment is loud in test and CI output:

import azure.functions as func

from azure_functions_langgraph import LangGraphApp

# Production (default): require function key authentication
app = LangGraphApp()  # equivalent to LangGraphApp(auth_level=func.AuthLevel.FUNCTION)

# Local dev only: explicit opt-in to anonymous — emits a UserWarning
app_local = LangGraphApp(auth_level=func.AuthLevel.ANONYMOUS)

Note: There are two health surfaces. The liveness probe GET /api/health uses health_auth_level, which defaults to ANONYMOUS independently of auth_level, and returns only {"status": "ok"} — it never enumerates registered graphs. Set health_auth_level=func.AuthLevel.FUNCTION to require a key on the probe as well.

The detailed inventory GET /api/health/details (graph names, descriptions, and checkpointer status) uses health_details_auth_level, which defaults to the app-level auth_level (FUNCTION) — so the inventory is protected by default. Pass health_details_auth_level=func.AuthLevel.ANONYMOUS explicitly to expose it publicly (e.g. local development).

Streaming behavior

Default (LangGraphApp): buffered SSE. All /stream endpoints on the classic LangGraphApp (the native POST /api/graphs/{name}/stream and the Platform-compatible POST /threads/{id}/runs/stream and POST /runs/stream) return buffered SSE. Chunks emitted by the graph are collected during execution and flushed as SSE events after the run completes — this is not true token-level streaming, and clients will not receive partial tokens incrementally. Buffered SSE is the classic adapter's deliberate implementation choice, because the classic HttpRequest/HttpResponse model it is built on cannot flush a response incrementally.

Opt-in true streaming: StreamingLangGraphApp (experimental). For genuine incremental delivery — each event: data frame flushed as the graph produces it — use the separate StreamingLangGraphApp. It serves the graph over the FastAPI/ASGI transport provided by the azurefunctions-extensions-http-fastapi extension (Azure Functions runtime 4.34.1+), gated behind the streaming extra:

pip install "azure-functions-langgraph[streaming]"

from azure_functions_langgraph import StreamingLangGraphApp

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

Enabling true streaming converts the entire function app to the ASGI model, so it cannot be mixed with classic LangGraphApp routes in the same app — deploy a separate app if you also need the classic buffered surface, and note the classic max_stream_response_bytes byte guard is replaced by max_stream_events (a frame count cap) because a true stream is never buffered. LangGraphApp is unchanged and remains the default. See examples/true_streaming_agent/ for a runnable curl -N walkthrough that proves incremental arrival.

Run observability

Wire a RunObserver to receive run-lifecycle signals (started / completed / failed / rejected) for every native invoke/stream run and every Platform runs/wait / runs/stream run. Each callback receives an immutable RunContext of correlation identifiers and timing only — never input, output, config, headers, or secrets.

The package ships a built-in, dependency-free LoggingRunObserver for zero-boilerplate telemetry:

import logging
from azure_functions_langgraph import LangGraphApp, LoggingRunObserver

logging.getLogger("azure_functions_langgraph.observability.run").setLevel(logging.INFO)

app = LangGraphApp(observer=LoggingRunObserver())
app.register(graph=graph, name="my_agent")

It emits one structured log record per lifecycle event under a single extra key (langgraph_run) — graph_name, endpoint, run_id, thread_id, assistant_id, stream_mode, transport, has_checkpointer, lock_backend, duration_ms, a derived status, and (on failure) the exception error_type (class name only). Azure Functions' Application Insights integration surfaces these under customDimensions.

The package owns the domain signal only — not log formatting, App Insights ingestion, OpenTelemetry exporters, sampling, or PII redaction. Observer failures are isolated and never fail a graph run. To write your own observer, see examples/run_observer/; for an App Insights + KQL walkthrough, see examples/observability_app_insights/.

OpenTelemetry observer (optional otel extra)

For distributed-tracing backends, the package ships an optional OTelRunObserver that maps each run's lifecycle to a single OpenTelemetry span. It is gated behind the otel extra:

pip install azure-functions-langgraph[otel]

from azure_functions_langgraph import LangGraphApp, OTelRunObserver

# Uses whatever TracerProvider the operator has already installed.
app = LangGraphApp(observer=OTelRunObserver())
app.register(graph=graph, name="my_agent")

on_run_started opens a span (keyed by run_id); the terminal event (completed / failed / rejected) closes it with the matching StatusCode, a langgraph.status, and a langgraph.duration_ms. Span attributes come only from the safe RunContext correlation fields — never input, output, config, headers, or secrets — and failures record the exception type only.

Consistent with the observability boundary above, OTelRunObserver does not configure a TracerProvider, exporters, sampling, or resource attributes — the operator owns all SDK/exporter wiring. The observer only acquires a tracer via opentelemetry.trace.get_tracer(...) and emits spans onto whatever provider is installed. For a runnable local demo with a console exporter, see examples/run_observer_otel/.

Service Bus trigger (experimental)

Beyond the HTTP surface, a compiled graph can be driven directly from an Azure Service Bus message via register_service_bus. Each delivered message is mapped to a single invoke (or ainvoke) call — useful for background jobs and decoupled pipelines where work arrives as messages rather than synchronous HTTP requests. A graph may be registered on both surfaces.

from azure_functions_langgraph import LangGraphApp

app = LangGraphApp()

# Queue trigger
app.register_service_bus(
    graph=graph,
    name="jobs_agent",
    connection="ServiceBusConnection",  # app-setting name
    queue_name="langgraph-jobs",
)

# Or a topic subscription trigger
app.register_service_bus(
    graph=graph,
    name="events_agent",
    connection="ServiceBusConnection",
    topic_name="events",
    subscription_name="langgraph",
)

func_app = app.function_app

HTTP vs Service Bus

Message mapping. By default the message body is decoded as UTF-8 and parsed as JSON: a JSON object is passed through as the graph input verbatim; any other value (or non-JSON text) is wrapped as {"messages": [{"role": "human", "content": <text>}]}. Pass input_mapper=... to customize.

Thread state and locking. Runs are threadless by default. Pass thread_id_factory=... to derive a checkpoint thread_id from a message; when it returns a value and the graph has a checkpointer, the app-level thread_lock guards the run so concurrent deliveries for the same logical thread do not mutate checkpoints concurrently. If the lock cannot be acquired a ThreadContentionError is raised so the binding abandons and retries.

Error handling and telemetry. Graph exceptions are not swallowed — a failing run propagates so the message is retried / dead-lettered per your queue's policy. This package ships no Service Bus output binding; pass a result_handler(result, msg) to publish results. Message bodies and application-property values are never included in telemetry; runs carry a trigger_type="service_bus" correlation field so they are distinguishable from HTTP runs in your observer. See examples/service_bus_agent/ for a full wiring.

Durable async run lifecycle (experimental)

For long-running graphs, opt into an experimental Durable Functions-backed async run control plane with LangGraphApp(async_runs="durable"). This adds a create / poll / cancel HTTP surface on top of the classic invoke/stream routes:

from azure_functions_langgraph import LangGraphApp

app = LangGraphApp(async_runs="durable")  # requires the `durable` extra
app.register(graph=graph, name="my_agent")
func_app = app.function_app

pip install "azure-functions-langgraph[durable]"

The graph executes inside a Durable activity, never the orchestrator, so Durable replay determinism is preserved — no LLM, tool, network, clock, random, or graph user code runs in the orchestrator. Durable history is not a replacement for LangGraph checkpoints; reuse a checkpointer for conversation state as usual. Durable dependencies stay optional behind the durable extra. See examples/durable_async_agent/ for a full create → poll → result → cancel walkthrough.

Per-graph auth

Override app-level auth settings per graph:

# Per-graph authentication override
app.register(graph=public_graph, name="public", auth_level=func.AuthLevel.ANONYMOUS)
app.register(graph=private_graph, name="private", auth_level=func.AuthLevel.FUNCTION)

Example request using a function key:

curl -X POST "https://<app>.azurewebsites.net/api/graphs/echo_agent/invoke?code=<FUNCTION_KEY>" \
  -H "Content-Type: application/json" \
  -d '{"input": {"messages": [{"role": "human", "content": "Hello!"}]}}'

Custom route prefix

All routes use the default /api prefix set by Azure Functions. To change it, configure routePrefix in your host.json:

{
  "extensions": {
    "http": {
      "routePrefix": "v1"
    }
  }
}

This changes all routes (e.g. POST /v1/graphs/{name}/invoke). Set routePrefix to "" to remove the prefix entirely.

Important — LangGraphApp(route_prefix=...) is metadata-only. Azure Functions resolves HTTP routes from host.json (the source of truth), not from the constructor argument. The route_prefix argument is recorded into the metadata snapshot consumed by tooling such as the azure-functions-openapi-python bridge so generated specs reflect the deployed routes, but changing it without also updating host.json does not change where requests are served. Always edit host.json to actually move routes, and pass the same value to LangGraphApp(route_prefix=...) so metadata stays in sync.

What you get

• POST /api/graphs/echo_agent/invoke — invoke the agent
• POST /api/graphs/echo_agent/stream — stream agent responses (buffered SSE; opt into true incremental streaming with StreamingLangGraphApp)
• GET /api/graphs/echo_agent/threads/{thread_id}/state — inspect thread state
• GET /api/health — liveness probe ({"status": "ok"})
• GET /api/health/details — registered-graph inventory (protected by default)

With platform_compat=True, you also get SDK-compatible endpoints:

• POST /assistants/search — list registered assistants
• GET /assistants/{id} — get assistant details
• POST /assistants/count — count assistants
• POST /threads — create thread
• GET /threads/{id} — get thread
• PATCH /threads/{id} — update thread metadata
• DELETE /threads/{id} — delete thread
• POST /threads/search — search threads
• POST /threads/count — count threads
• POST /threads/{id}/runs/wait — run and wait for result
• POST /threads/{id}/runs/stream — run and stream result (buffered SSE)
• POST /runs/wait — threadless run
• POST /runs/stream — threadless stream (buffered SSE)
• GET /threads/{id}/state — get thread state
• POST /threads/{id}/state — update thread state
• POST /threads/{id}/history — get state history

Request format

{
    "input": {
        "messages": [{"role": "human", "content": "Hello!"}]
    },
    "config": {
        "configurable": {"thread_id": "conversation-1"}
    }
}

Persistent storage (v0.4+)

Use Azure Blob Storage for checkpoint persistence and Azure Table Storage for thread metadata:

import azure.functions as func

from azure.storage.blob import ContainerClient
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

from azure_functions_langgraph import LangGraphApp
from azure_functions_langgraph.checkpointers.azure_blob import AzureBlobCheckpointSaver
from azure_functions_langgraph.stores.azure_table import AzureTableThreadStore


class AgentState(TypedDict):
    messages: list[dict[str, str]]


def chat(state: AgentState) -> dict:
    user_msg = state["messages"][-1]["content"]
    return {"messages": state["messages"] + [{"role": "assistant", "content": f"Echo: {user_msg}"}]}


# Build graph with Azure Blob checkpointer
container_client = ContainerClient.from_connection_string(
    "DefaultEndpointsProtocol=https;AccountName=...", "checkpoints"
)
saver = AzureBlobCheckpointSaver(container_client=container_client)

builder = StateGraph(AgentState)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
graph = builder.compile(checkpointer=saver)

# Deploy with Azure Table thread store
thread_store = AzureTableThreadStore.from_connection_string(
    "DefaultEndpointsProtocol=https;AccountName=...", table_name="threads"
)

# Production: always set auth_level explicitly
app = LangGraphApp(platform_compat=True, auth_level=func.AuthLevel.FUNCTION)
app.thread_store = thread_store
app.register(graph=graph, name="echo_agent")
func_app = app.function_app

Checkpoints and thread metadata survive Azure Functions restarts and scale across instances.

Persistent storage with Managed Identity

The recommended production wiring uses Managed Identity instead of connection strings, so no secrets land in App Settings. Install the azure-identity extra and pass DefaultAzureCredential to both clients:

pip install azure-functions-langgraph[azure-blob,azure-table,azure-identity]

from azure.data.tables import TableClient
from azure.identity import DefaultAzureCredential
from azure.storage.blob import ContainerClient

from azure_functions_langgraph.checkpointers.azure_blob import AzureBlobCheckpointSaver
from azure_functions_langgraph.stores.azure_table import AzureTableThreadStore

credential = DefaultAzureCredential()

container_client = ContainerClient(
    account_url="https://<account>.blob.core.windows.net",
    container_name="langgraph-checkpoints",
    credential=credential,
)
table_client = TableClient(
    endpoint="https://<account>.table.core.windows.net",
    table_name="langgraphthreads",
    credential=credential,
)

checkpointer = AzureBlobCheckpointSaver(container_client=container_client)
thread_store = AzureTableThreadStore.from_table_client(table_client=table_client)

Required role assignments on the storage account (or narrower scopes):

DefaultAzureCredential walks a chain of credentials. In Azure Functions it picks up the Function App's Managed Identity; locally it falls back to AzureCliCredential (az login) — the same code path works in both environments without conditional wiring.

For a complete runnable example (Managed Identity in prod, Azurite + connection string locally), see examples/managed_identity_storage/.

Native async checkpoint I/O

By default AzureBlobCheckpointSaver is configured with a synchronous ContainerClient. Its async methods (aget_tuple, alist, aput, aput_writes, adelete_thread) remain fully available: when a graph is driven through the async endpoints (graph.ainvoke / graph.astream) without an aio client, they run the synchronous implementation in a thread executor and emit a one-time warning — correct, but not truly non-blocking.

To get native, non-blocking blob I/O on the async path, pass an azure.storage.blob.aio.ContainerClient as aio_container_client alongside the synchronous container_client. Both clients share the same container; the sync methods use the sync client and the async methods use the aio client, so checkpoints written by one path are readable by the other:

from azure.identity import DefaultAzureCredential
from azure.identity.aio import DefaultAzureCredential as AioDefaultAzureCredential
from azure.storage.blob import ContainerClient
from azure.storage.blob.aio import ContainerClient as AioContainerClient

container_client = ContainerClient(
    account_url="https://<account>.blob.core.windows.net",
    container_name="langgraph-checkpoints",
    credential=DefaultAzureCredential(),
)
aio_container_client = AioContainerClient(
    account_url="https://<account>.blob.core.windows.net",
    container_name="langgraph-checkpoints",
    credential=AioDefaultAzureCredential(),
)

checkpointer = AzureBlobCheckpointSaver(
    container_client=container_client,
    aio_container_client=aio_container_client,
)

The caller owns the lifecycle of both clients (open/close), exactly as with the synchronous client. aio_container_client is fully optional and backward-compatible — omitting it preserves the previous executor-fallback behaviour.

Checkpoint store security

The checkpointer backends persist graph state using LangGraph's default serializer, which can restore arbitrary Python types from a checkpoint payload. Treat the checkpoint store as part of your threat model — checkpoint blobs are not trusted-free data:

• Restrict storage access with RBAC / Managed Identity (above) and private endpoints so only the Function App identity can read or write checkpoints.
• Enable strict deserialization by setting LANGGRAPH_STRICT_MSGPACK=true (plus any allowed modules your graphs need) in your Function App application settings. The env var is read at import time, so it must be set before the app imports LangGraph — configure it as an app setting, not at runtime. Upstream defaults to permissive (false).

See the upstream advisory GHSA-g48c-2wqr-h844 and docs/security.md for the full rationale.

Scale envelope

The bundled persistent backends are intended for development and small-to-medium production deployments. Plan ahead before pushing past these limits:

Notes:

• Single partition — AzureTableThreadStore keys every thread under a single PartitionKey, capped by Azure Table per-partition throughput (~2000 entities/sec on Standard accounts). Search and count beyond status filtering are client-side; see COMPATIBILITY.md.
• Prefix scans — AzureBlobCheckpointSaver lists checkpoints via blob prefix scans; transaction count and latency grow with checkpoints-per-thread. Use the retention helpers below to keep that bounded.
• Entity size — Azure Table entities are capped at 1 MB; the store logs a warning at 90% of the threshold.

Retention helpers

AzureBlobCheckpointSaver exposes two helpers for scheduled cleanup (e.g. from a Timer-triggered Function):

# Keep only the most recent 50 checkpoints per (thread, namespace)
saver.delete_old_checkpoints(thread_id="conversation-1", keep_last=50)

# Or delete everything older than a known checkpoint id
saver.delete_checkpoints_before(
    thread_id="conversation-1",
    before_checkpoint_id="01HXY...",
)

Both helpers only delete checkpoint marker, metadata, and write blobs. They intentionally preserve channel value blobs (under values/) and the latest.json pointer so retained checkpoints remain fully usable.

Note — delete_old_checkpoints / delete_checkpoints_before are safe but not exhaustive. Channel value blobs that were referenced only by the now-deleted checkpoints become orphaned and are not removed. For long-running threads with frequent checkpointing, those orphans can dominate the storage footprint over time. Run collect_orphaned_values() (below) on a schedule as the second step.

Garbage-collecting orphaned channel values

After pruning checkpoints, collect_orphaned_values() walks the surviving checkpoints, builds the set of (channel, version) pairs they reference, and removes any values/ blob outside that set. Default is dry-run so you can audit first:

# Dry run — see what would be deleted, change nothing
audit = saver.collect_orphaned_values(thread_id="conversation-1")
print(audit.would_delete)

# Real run — actually delete orphans
result = saver.collect_orphaned_values(thread_id="conversation-1", dry_run=False)
print(f"Deleted {len(result.deleted)} orphaned value blobs")

The helper is concurrency-safe by two complementary mechanisms:

1. Recent-write grace period — value blobs whose last_modified is within grace_period_seconds (default 300s) are deferred to a future GC pass and recorded in result.skipped_recent. This protects the window between a value blob being uploaded and its checkpoint commit marker (latest.json) being finalized.
2. Per-orphan re-scan — immediately before each delete, the survivor set is recomputed; an older value blob that a newly finalized checkpoint started referencing after the snapshot is preserved (such blobs appear in would_delete but not deleted).

Per namespace, the helper fails closed — if latest.json is missing or any surviving checkpoint blob is unreadable / fails deserialization, the namespace is skipped entirely (recorded in result.skipped_namespaces) so a misconfigured or transiently-unavailable store cannot trigger destructive deletion.

Set grace_period_seconds=0 to disable the recent-write guard during an offline maintenance window when no concurrent checkpoint writes are possible.

DB checkpointer backends

For workloads that already run a managed database (or need state shared across multiple Function instances), thin DX helpers wrap the official LangGraph DB checkpoint packages without reimplementing storage:

Each helper owns the connection lifetime and emits clear ImportErrors pointing at the right extra. The Postgres and SQLite helpers accept a connection string and (by default) call upstream setup() on cold start so the checkpoint tables exist; the Cosmos DB helper accepts an endpoint and key, temporarily wires the upstream COSMOSDB_ENDPOINT / COSMOSDB_KEY environment variables, and directly instantiates the upstream CosmosDBSaver:

Authentication — create_cosmos_checkpointer is a key-based convenience wrapper; Managed Identity is available upstream. The DX helper resolves an account key and temporarily wires COSMOSDB_ENDPOINT / COSMOSDB_KEY before instantiating the upstream CosmosDBSaver, so calling the helper always uses key-based auth. However, the upstream langgraph-checkpoint-cosmosdb package (≥ 0.2.8) does support passwordless auth: CosmosDBSaver falls back to DefaultAzureCredential (Managed Identity, az login, service principal) whenever COSMOSDB_KEY is unset. To use Managed Identity today, skip the helper and instantiate the upstream saver directly with only COSMOSDB_ENDPOINT set (no key):

import os
from langgraph_checkpoint_cosmosdb import CosmosDBSaver

os.environ["COSMOSDB_ENDPOINT"] = "https://<account>.documents.azure.com:443/"
# Leave COSMOSDB_KEY unset → upstream uses DefaultAzureCredential (Managed Identity)
checkpointer = CosmosDBSaver(database_name="langgraph", container_name="checkpoints")

Grant the Function App's identity a Cosmos DB data-plane role (e.g. Cosmos DB Built-in Data Contributor) on the account. The official langchain-azure-cosmosdb package is an alternative that also supports DefaultAzureCredential. A future release may extend create_cosmos_checkpointer with a first-class Managed Identity path; until then the helper remains key-based by design.

import os
from azure_functions_langgraph.checkpointers.postgres import create_postgres_checkpointer

checkpointer = create_postgres_checkpointer(
    os.environ["LANGGRAPH_POSTGRES_CONNECTION_STRING"],
    setup=True,  # set False once your deployment pipeline owns migrations
)
graph = builder.compile(checkpointer=checkpointer)

The helpers do not hide builder.compile(checkpointer=...) and do not reimplement DB checkpoint storage — they centralize connection conventions and emit clear ImportErrors pointing at the right extra. The Postgres and SQLite helpers run setup() once at cold start; the Cosmos DB helper directly instantiates the saver (no setup() call, no context manager). See examples/postgres_checkpoint_production/, examples/sqlite_checkpoint_local/, and examples/cosmos_checkpoint_azure/ for full Azure-Functions wiring.

Run lock semantics

When using AzureTableThreadStore with Platform-compatible runs, each graph execution acquires an atomic run lock (ETag compare-and-swap) on the thread before invoking the graph. On completion — success or failure — the lock is released via a best-effort merge update.

What can go wrong: If a Function host instance is terminated during graph execution (scale-in, deployment, crash), the thread remains in busy status indefinitely because the release never fires.

Recovery: Use reset_stale_locks() from a periodic Timer Trigger to reclaim orphaned locks:

# Reset threads stuck in 'busy' for more than 10 minutes
count = thread_store.reset_stale_locks(older_than_seconds=600)

Each reset uses ETag CAS so a thread that has been legitimately re-acquired since the scan is never stomped. Choose older_than_seconds comfortably above your longest expected graph execution time — ETag CAS protects against re-acquire races, but a still-running long job will not update its updated_at and could be reclaimed if the threshold is too short. See examples/maintenance_timer/ for a complete Timer Trigger wiring.

Stale-lock cleanup caveat: reset_stale_locks issues a projection query (select=["RowKey", "updated_at"]) and relies on the Azure Tables SDK to expose each row's ETag via either entity.metadata["etag"] or entity["etag"]. Rows where neither shape populates an ETag are skipped (logged at DEBUG) so a stale lock is never reset without a writable ETag for CAS; such rows are retried on the next scan once the SDK returns a usable ETag.

Native endpoint thread locking

Native invoke/stream endpoints (POST /api/graphs/{name}/invoke and .../stream) use an in-process per-thread lock when the graph has a checkpointer and the request includes config.configurable.thread_id. This prevents concurrent writes to single-writer checkpointers (e.g. AzureBlobCheckpointSaver) within the same Python worker process.

Important: The default backend is in-process only — it does not coordinate across multiple Function App instances, worker processes, or hosts. Multi-instance deployments (Consumption / Elastic Premium) must either use a distributed ThreadLock backend (see below) or route through Platform-compatible runs (platform_compat=True) with AzureTableThreadStore, which provides ETag-based atomic locking.

Distributed thread locking (v0.6+)

The thread_lock parameter accepts any object satisfying the ThreadLock protocol, so multi-instance deployments can swap the default InProcessThreadLock for a distributed backend. The package ships with AzureBlobLeaseThreadLock, which uses Azure Blob lease compare-and-swap as the underlying primitive — no additional infra beyond a Blob container.

import azure.functions as func
from azure.storage.blob import ContainerClient

from azure_functions_langgraph import LangGraphApp
from azure_functions_langgraph.locks import AzureBlobLeaseThreadLock

container = ContainerClient.from_connection_string(
    "DefaultEndpointsProtocol=https;AccountName=...", "langgraph-locks"
)
container.create_container()  # idempotent — safe on cold start

app = LangGraphApp(
    auth_level=func.AuthLevel.FUNCTION,
    thread_lock=AzureBlobLeaseThreadLock(container_client=container),
)

Safety guard — AZFUNC_LANGGRAPH_LOCK_BACKEND. Set this environment variable to distributed in multi-instance environments to fail-fast at startup if a distributed backend was not wired. When the value is distributed (or any non-empty value other than inprocess) and thread_lock resolves to the default InProcessThreadLock, LangGraphApp.__post_init__ raises RuntimeError — this prevents accidentally deploying an in-process lock to a horizontally-scaled Function App. See docs/production-guide.md for the full scale-out matrix.

Upgrading

v0.3.0 → v0.4.0

Fully backward-compatible. No breaking changes.

• New optional extras: pip install azure-functions-langgraph[azure-blob,azure-table] for persistent storage
• New platform endpoints: thread CRUD, state update/history, threadless runs, assistants count
• New protocols: UpdatableStateGraph, StateHistoryGraph (available from azure_functions_langgraph.protocols)

v0.4.0 → v0.5.0

Fully backward-compatible. No breaking changes.

• Metadata API: app.get_app_metadata() returns an immutable snapshot of all registered routes and graph info
• OpenAPI bridge: azure_functions_langgraph.openapi.register_with_openapi integrates with azure-functions-openapi-python
• CloneableGraph protocol: thread-isolated graph cloning for safe concurrent execution

When to use

• You have LangGraph agents and want to deploy them on Azure Functions
• You want serverless deployment without LangGraph Platform costs
• You need HTTP endpoints for your compiled graphs with minimal setup
• You want thread-based conversation state via LangGraph checkpointers
• You need durable state persistence with Azure Blob/Table Storage

Documentation

• Start here: 5-minute tutorial — compile a graph → expose as HTTP → run locally → deploy to Azure
• Project docs live under docs/
• Smoke-tested examples live under examples/
• Product requirements: PRD.md
• Design principles: DESIGN.md

Ecosystem

This package is part of the Azure Functions Python DX Toolkit.

Design principle: azure-functions-langgraph owns LangGraph runtime exposure. azure-functions-validation-python owns validation. azure-functions-openapi-python owns API documentation.

For AI Coding Assistants

If you are an AI coding assistant (Copilot, Cursor, Claude, etc.), see:

• llms.txt — Concise package summary and API overview
• llms-full.txt — Complete API reference with signatures, patterns, and examples

Disclaimer

This project is an independent community project and is not affiliated with, endorsed by, or maintained by Microsoft or LangChain.

Azure and Azure Functions are trademarks of Microsoft Corporation. LangGraph and LangChain are trademarks of LangChain, Inc.

License

MIT