How to Integrate LangGraph AI Agents with ASP.NET Core 10 Using C# and Python (2026)

A workflow that pauses for human approval has to remember where it stopped. Keeping the workflow in a web request until someone answers ties recovery to a live connection and can accidentally repeat work when a request is retried.

This guide connects ASP.NET Core 10 to a Python LangGraph workflow with SQLite checkpoints. The graph prepares a preview, routes longer text to an approval step, and resumes from the same thread. The deterministic nodes make pause, reject, and replay behavior easy to test without a model provider.

What does LangGraph add?

The graph prepares a preview, interrupts for approval when needed, and resumes using the same thread checkpoint.

LangGraph models state, nodes, edges and persistence. LangChain provides components such as retrievers and model integrations that can run inside nodes. Microsoft Agent Framework offers a .NET-native choice for agent orchestration. This example uses Python because its checkpointed graph already owns the workflow. The .NET gateway exposes only start and decision routes.

Feature	Role in this example
Node	Prepare, pause for approval, or finish
Conditional edge	Choose approval or direct completion
Checkpointer	Persist graph state by thread ID
Interrupt and resume	Return a preview, then continue after a decision

The interrupted node may restart from its beginning when resumed. Keep side effects before an interrupt idempotent, or move them to a later node with a durable execution record. This lab finishes by setting a state string. It sends no message and writes no external business object. See LangGraph interrupts and Microsoft Agent Framework workflows.

Build the graph service and C# bridge

Install Python 3.12 and .NET 10. Put the following files in the same directory. Configure a shared random bridge key and keep the worker port private. The approval flag in the lab is provided by a trusted local operator. In production the API must verify an approver identity and check ownership of the exact thread before forwarding any decision.

requirements.txt

fastapi==0.115.12
uvicorn==0.34.2
langgraph==1.0.5
langgraph-checkpoint-sqlite==3.0.1

service.py

import hmac
import os
import re
import sqlite3
from threading import Lock
from typing import TypedDict

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from pydantic import BaseModel, Field

key = os.environ["BridgeKey"]
if len(key) < 32:
    raise RuntimeError("Use a random BridgeKey.")
app = FastAPI()
lock = Lock()
db = sqlite3.connect(os.getenv("GraphDb", "checkpoints.sqlite"), check_same_thread=False)


@app.middleware("http")
async def authenticate(request: Request, call_next):
    if not hmac.compare_digest(request.headers.get("X-Bridge-Key", "").encode(), key.encode()):
        return JSONResponse({"detail": "Unauthorized"}, status_code=401)
    return await call_next(request)


class State(TypedDict):
    text: str
    status: str


def prepare(state: State):
    return {"text": state["text"].strip(), "status": "prepared"}


def route(state: State):
    return "approve" if len(state["text"]) > 20 else "finish"


def approve(state: State):
    decision = interrupt({"question": "Approve this preview?", "preview": state["text"]})
    return {"status": "approved" if decision is True else "rejected"}


def finish(state: State):
    return {"status": "completed" if state["status"] != "rejected" else "rejected"}


builder = StateGraph(State)
builder.add_node("prepare", prepare)
builder.add_node("approve", approve)
builder.add_node("finish", finish)
builder.add_edge(START, "prepare")
builder.add_conditional_edges("prepare", route, {"approve": "approve", "finish": "finish"})
builder.add_edge("approve", "finish")
builder.add_edge("finish", END)
graph = builder.compile(checkpointer=SqliteSaver(db))


class Start(BaseModel):
    text: str = Field(min_length=1, max_length=500)


class Decision(BaseModel):
    approved: bool


def config(thread: str):
    if not re.fullmatch(r"[a-f0-9-]{36}", thread):
        raise HTTPException(400, "Use a UUID thread ID")
    return {"configurable": {"thread_id": thread}}


def result(data):
    if pending := data.get("__interrupt__"):
        return {"status": "pending", "approval": pending[0].value}
    return {"status": data["status"], "preview": data["text"]}


@app.post("/start/{thread}")
def start(thread: str, body: Start):
    cfg = config(thread)
    with lock:
        if graph.get_state(cfg).values:
            raise HTTPException(409, "Thread already exists")
        return result(graph.invoke({"text": body.text, "status": "new"}, cfg))


@app.post("/resume/{thread}")
def resume(thread: str, body: Decision):
    cfg = config(thread)
    with lock:
        state = graph.get_state(cfg)
        if not state.values or not any(task.interrupts for task in state.tasks):
            raise HTTPException(409, "No pending approval")
        return result(graph.invoke(Command(resume=body.approved), cfg))

Lab.csproj

<Project Sdk="Microsoft.NET.Sdk.Web">
  <PropertyGroup><TargetFramework>net10.0</TargetFramework><ImplicitUsings>enable</ImplicitUsings><Nullable>enable</Nullable></PropertyGroup>
</Project>

Program.cs

using System.Text.Json;
var b = WebApplication.CreateBuilder(args);
var key = b.Configuration["BridgeKey"] ?? throw new InvalidOperationException("Set BridgeKey.");
if (key.Length < 32) throw new InvalidOperationException("Use a random BridgeKey.");
b.WebHost.ConfigureKestrel(o => o.Limits.MaxRequestBodySize = 32768);
b.Services.AddHttpClient("worker", h =>
{
    h.BaseAddress = new Uri("http://127.0.0.1:8000/");
    h.Timeout = TimeSpan.FromSeconds(15);
    h.DefaultRequestHeaders.Add("X-Bridge-Key", key);
}).ConfigurePrimaryHttpMessageHandler(() => new HttpClientHandler { AllowAutoRedirect = false });
var app = b.Build();
app.Use(async (c, next) =>
{
    if (c.Request.Host.Host != "127.0.0.1" || c.Request.Headers.ContainsKey("Origin"))
    { c.Response.StatusCode = 403; return; }
    await next(c);
});
app.MapPost("/runs/{id:guid}", (Guid id, JsonElement p, IHttpClientFactory f, CancellationToken ct) => Forward($"start/{id:D}", p, f, ct));
app.MapPost("/runs/{id:guid}/decision", (Guid id, JsonElement p, IHttpClientFactory f, CancellationToken ct) => Forward($"resume/{id:D}", p, f, ct));
await app.RunAsync();

static async Task<IResult> Forward(string path, JsonElement input, IHttpClientFactory clients, CancellationToken ct)
{
    try
    {
        using var http = clients.CreateClient("worker");
        using var reply = await http.PostAsJsonAsync(path, input, ct);
        if ((int)reply.StatusCode is 400 or 409 or 422 or 429)
            return Results.StatusCode((int)reply.StatusCode);
        if (!reply.IsSuccessStatusCode) return Results.Problem(statusCode: 502);
        using var json = await JsonDocument.ParseAsync(await reply.Content.ReadAsStreamAsync(ct),
            cancellationToken: ct);
        return Results.Json(json.RootElement.Clone());
    }
    catch (OperationCanceledException) when (!ct.IsCancellationRequested)
    { return Results.Problem(statusCode: 504); }
    catch (Exception e) when (e is HttpRequestException or JsonException)
    { return Results.Problem(statusCode: 502); }
}

The worker serializes local changes with a process lock and uses a SQLite checkpointer. Run a single worker against this fixture. The checkpointer stores the state but does not authorize who may read or resume it. A thread UUID is not a credential. For multiple replicas, adopt a supported shared checkpointer and a concurrency policy across processes.

Test pause, resume, and replay

Start the Python worker

python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
export BridgeKey="$(openssl rand -hex 32)"
uvicorn service:app --host 127.0.0.1 --port 8000

Set the same key in the .NET terminal and run the API. Use a fresh thread UUID per test.

Start a graph and approve it

dotnet run --no-launch-profile -- --urls http://127.0.0.1:5080

# In a separate terminal:
id=11111111-1111-1111-1111-111111111111
curl -i "http://127.0.0.1:5080/runs/$id" -H 'Content-Type: application/json' \
  -d '{"text":"Please review this longer support draft before it is used."}'
curl -i "http://127.0.0.1:5080/runs/$id/decision" -H 'Content-Type: application/json' \
  -d '{"approved":true}'

Expect pending with a preview, then completed. Send the same decision again and expect 409. Try a separate thread with false and expect rejected. For a recovery test, restart the Python process while the workflow is pending and use the same database path and graph definition before resuming. A different thread ID should not resume the original graph.

The C# project compiled. A loopback test returned pending with a preview, resumed to completed, and rejected a duplicate decision with 409. Restart recovery, rejection and production identity were not exercised.

Add safe operational boundaries

Scope start and resume by tenant, caller, workflow revision and requested action. Persist the approval payload or a canonical hash and compare it before any side effect. A boolean decision is sufficient for this preview lab, but an approval for one payload must not authorize a changed payload. Apply a decision deadline and use a separate cancellation state for abandoned runs. Use immutable audit records for meaningful actions and avoid logging sensitive preview text.

A restart with incompatible node names or state serialization can break resumption. Version the graph and retain a migration or draining plan. A checkpoint lets a workflow recover control flow. It does not guarantee exactly-once external actions. For a reliable side effect, write an execution intent transactionally and let a worker use a destination-supported idempotency key.

Symptom	First check
Resume returns 409	Correct thread ID and pending interrupt
Pending preview lost after restart	Database path and durable volume
Same action executes twice	Side effect before interrupt or missing dedupe
C# returns 502	Worker process, bridge authentication or response format
More replicas create races	Process-local lock is not distributed

FAQ and references

Can I use LangGraph without an LLM? Yes. Nodes can run deterministic code. Model calls can be added where they improve a specific decision, with explicit validation.

Does approval resume a live request? The original request already returned. A later request supplies a resume value for the persisted thread.

Can I hand off to another framework? Keep the boundary as an authenticated contract with versioned state or messages. Avoid sharing opaque internal objects across runtimes.

Make the approval transition observable before adding agents with side effects. A deterministic fixture exposes replay and recovery mistakes early.

Search This Blog

ASP.NET PB – ASP.NET Core, C#, SQL Server & AI Tutorials