Deploy a LangGraph agent

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import interrupt

# Stand-in data: replace with calls to your order system.
ORDERS = {"1001": "shipped", "1002": "delivered"}


@tool
def get_order_status(order_id: str) -> str:
    """Look up an order's status by its order number."""
    return ORDERS.get(order_id, "not found")


@tool
def refund_order(order_id: str, reason: str) -> str:
    """Refund an order. A staff member approves every refund before it's issued."""
    decision = interrupt({"action": "refund_order", "order_id": order_id, "reason": reason})
    if decision != "approve":
        return f"A staff member declined the refund for order {order_id}."
    # Issue the refund with your payment provider here.
    return f"A staff member approved the refund. Order {order_id} was refunded."


tools = [get_order_status, refund_order]
# One tool call at a time, so a thread waits for one decision at a time.
model = ChatAnthropic(model="claude-haiku-5-5").bind_tools(tools, parallel_tool_calls=False)
system = SystemMessage("You answer questions about Acme Storefront orders and refund orders when customers ask.")


async def agent(state: MessagesState):
    return {"messages": [await model.ainvoke([system, *state["messages"]])]}


builder = StateGraph(MessagesState)
builder.add_node(agent)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")

import asyncio
import os

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver


async def main() -> None:
    async with AsyncPostgresSaver.from_conn_string(os.environ["DB_URL"]) as checkpointer:
        await checkpointer.setup()


asyncio.run(main())

import os
import secrets
from contextlib import asynccontextmanager
from typing import Literal

from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi.sse import EventSourceResponse, ServerSentEvent
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.types import Command
from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from pydantic import BaseModel

from app.graph import builder

API_TOKEN = os.environ["API_TOKEN"].encode()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # The settings AsyncPostgresSaver needs on every connection.
    kwargs = {"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row}
    async with AsyncConnectionPool(os.environ["DB_URL"], kwargs=kwargs) as pool:
        app.state.graph = builder.compile(checkpointer=AsyncPostgresSaver(pool))
        yield


def check_token(credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
    if not secrets.compare_digest(credentials.credentials.encode(), API_TOKEN):
        raise HTTPException(status_code=401, detail="Invalid token")


app = FastAPI(lifespan=lifespan, dependencies=[Depends(check_token)])


def thread(thread_id: str) -> dict:
    return {"configurable": {"thread_id": thread_id}}


async def thread_state(thread_id: str, request: Request):
    graph = request.app.state.graph
    state = await graph.aget_state(thread(thread_id))
    if state.next and not state.interrupts:
        # A deployment stopped the last run midway: finish its remaining steps first.
        await graph.ainvoke(None, thread(thread_id), durability="sync")
        state = await graph.aget_state(thread(thread_id))
    return state


def not_waiting(state=Depends(thread_state)):
    if state.interrupts:
        raise HTTPException(status_code=409, detail="The thread is waiting for a decision")


def waiting(state=Depends(thread_state)):
    if not state.interrupts:
        raise HTTPException(status_code=409, detail="The thread has no decision to make")


async def stream(graph, input, thread_id: str):
    # durability="sync" saves each step before the next one starts.
    async for mode, data in graph.astream(
        input, thread(thread_id), stream_mode=["messages", "updates"], durability="sync"
    ):
        if mode == "messages":
            chunk, metadata = data
            if metadata["langgraph_node"] == "agent" and chunk.text:
                yield ServerSentEvent(event="delta", data=chunk.text)
        elif "__interrupt__" in data:
            yield ServerSentEvent(event="interrupt", data=data["__interrupt__"][0].value)
        elif "agent" in data:
            for call in data["agent"]["messages"][-1].tool_calls:
                yield ServerSentEvent(event="tool", data=call["name"])
    yield ServerSentEvent(event="done")


class Message(BaseModel):
    message: str


class Decision(BaseModel):
    decision: Literal["approve", "reject"]


@app.post("/threads/{thread_id}/messages", response_class=EventSourceResponse, dependencies=[Depends(not_waiting)])
async def send_message(thread_id: str, body: Message, request: Request):
    async for event in stream(request.app.state.graph, {"messages": [("user", body.message)]}, thread_id):
        yield event


@app.post("/threads/{thread_id}/resume", response_class=EventSourceResponse, dependencies=[Depends(waiting)])
async def resume(thread_id: str, body: Decision, request: Request):
    async for event in stream(request.app.state.graph, Command(resume=body.decision), thread_id):
        yield event


@app.get("/threads/{thread_id}")
async def get_thread(thread_id: str, request: Request):
    state = await request.app.state.graph.aget_state(thread(thread_id))
    return {
        "messages": [{"type": m.type, "text": m.text} for m in state.values.get("messages", []) if m.text],
        "waiting_for": state.interrupts[0].value if state.interrupts else None,
    }

FROM python:3.14-slim-trixie AS build
COPY --from=ghcr.io/astral-sh/uv:0.13 /uv /usr/local/bin/
ENV UV_COMPILE_BYTECODE=1 \
    UV_PYTHON_DOWNLOADS=0
WORKDIR /app
COPY pyproject.toml uv.lock .python-version ./
RUN uv sync --locked --no-dev

FROM python:3.14-slim-trixie
RUN useradd --uid 1000 --create-home app
ENV PATH="/app/.venv/bin:$PATH" \
    PYTHONUNBUFFERED=1 \
    PORT=8080 \
    FORWARDED_ALLOW_IPS="*"
WORKDIR /app
COPY --from=build /app/.venv .venv
COPY . .
USER 1000:1000
EXPOSE 8080
CMD ["sh", "-c", "exec uvicorn app.main:app --host 0.0.0.0 --port \"$PORT\""]

.git
.venv
**/__pycache__
.env*
Dockerfile
.dockerignore