Use when productionising or deploying a LangChain / LangGraph / DeepAgents agent. Covers durable execution (checkpointers, thread_id), the production middleware stack, three deploy targets (LangSmith Cloud, Cloud Run, Docker), secrets, scaling, and post-deploy verification.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Use when productionising or deploying a LangChain / LangGraph / DeepAgents agent. Covers durable execution (checkpointers, thread_id), the production middleware stack, three deploy targets (LangSmith Cloud, Cloud Run, Docker), secrets, scaling, and post-deploy verification.
Deploy + Productionisation
This skill has two halves: productionisation (what to put in the agent BEFORE you deploy it anywhere) and deploy (how to ship it).
Part 1: Productionisation
A "production-ready" agent has three things any toy agent does not:
Durable execution โ a checkpointer + thread_id so state survives restarts and interrupt() works.
The production middleware stack โ call limits, retries, fallbacks, summarization, PII handling, optional HITL.
Smoke evals you actually run before each deploy.
Durable execution
A graph becomes durable by attaching a checkpointer at compile time. Without one, interrupt() and HumanInTheLoopMiddleware do nothing useful, and crash recovery is impossible.
from langgraph.checkpoint.postgres import PostgresSaver
agent = create_agent(
model="claude-sonnet-4-6",
tools=[...],
middleware=[...], # see below
checkpointer=PostgresSaver.from_conn_string("postgresql://..."),
)
Checkpointer
When
InMemorySaver
Dev, tests, smoke runs. State dies with the process.
result = agent.invoke(
{"messages": [...]},
config={"configurable": {"thread_id": f"user-{user_id}-conv-{conv_id}"}},
)
To resume an interrupted thread (HITL approval, crash recovery), pass None as the input with the same thread_id:
result = agent.invoke(None, config={"configurable": {"thread_id": "user-42-conv-7"}})
Cleanup: old checkpoints accumulate. Either set up a job that deletes rows older than N days from the checkpoint table, or run with a TTL strategy at the app layer (drop threads older than X). The LangChain docs do not ship a built-in cleaner; this is your job.
The production middleware stack
Copy this and tune. See the langchain-agents-middleware skill for full details on each.
from langchain.agents import create_agent
from langchain.agents.middleware import (
ModelCallLimitMiddleware, ToolCallLimitMiddleware,
ModelRetryMiddleware, ToolRetryMiddleware,
ModelFallbackMiddleware, SummarizationMiddleware,
HumanInTheLoopMiddleware, PIIMiddleware,
)
from langgraph.checkpoint.postgres import PostgresSaver
agent = create_agent(
model="claude-sonnet-4-6",
tools=TOOLS,
middleware=[
# Order matters: limits BEFORE retries (so retries can't blow the budget)
ModelCallLimitMiddleware(run_limit=50),
ToolCallLimitMiddleware(run_limit=200),
# Resilience to transient failures
ModelRetryMiddleware(max_retries=3, backoff_factor=2.0),
ToolRetryMiddleware(max_retries=3, backoff_factor=2.0),
# Provider-level resilience
ModelFallbackMiddleware("openai:gpt-4o-mini"),
# Long-conversation hygiene
SummarizationMiddleware(model="claude-haiku-4-5", trigger=("tokens", 8000), keep=("messages", 20)),
# HITL on irreversible tools (requires checkpointer; below)
HumanInTheLoopMiddleware(interrupt_on={
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
"charge_card": {"allowed_decisions": ["approve", "reject"]},
}),
# Privacy (only if input may contain PII)
PIIMiddleware("email", strategy="redact", apply_to_input=True),
],
checkpointer=PostgresSaver.from_conn_string(os.environ["POSTGRES_URL"]),
)
Cost controls beyond ModelCallLimitMiddleware
Pick a small fallback.ModelFallbackMiddleware("openai:gpt-4o-mini") after a strong primary keeps costs bounded on retries.
Use LLMToolSelectorMiddleware when the agent has 10+ tools. A small model picks the relevant 3โ5 to expose to the main model, dropping prompt tokens.
Use SummarizationMiddleware on long conversations. Summarize every N tokens to keep prompt size bounded.
Use ContextEditingMiddleware to drop old tool outputs from context once they're no longer useful.
Structured outputs
If the agent's final answer must be typed (an extracted record, a decision), use model.with_structured_output(...)as the model passed to create_agent:
langgraph deploy pushes to LangSmith Cloud (managed Agent Server) and prints the deployment URL. State persistence and durable execution are managed for you โ you don't need to set up Postgres yourself; LangSmith provides it.
For secrets: set them in the LangSmith UI under the deployment's settings, or langgraph deploy --env KEY=value (one flag per secret โ securely stored).
For scaling: LangSmith handles horizontal scaling; you configure concurrency / min-instances in the UI.
Target 2: Google Cloud Run
gcloud run deploy --source . does it all: Cloud Build builds the image (using the project's Dockerfile if present), pushes to Artifact Registry, deploys the service. No local Docker needed.
Prerequisites
# Authenticated
gcloud auth list --filter=status:ACTIVE --format="value(account)"# Project + region
gcloud config get-value project
gcloud config get-value compute/region
# Required APIs
gcloud services list --enabled --filter="config.name:(run.googleapis.com OR cloudbuild.googleapis.com OR secretmanager.googleapis.com)" --format="value(config.name)"
For each KEY=value in .env, ensure a Secret Manager secret exists. Naming: <service>-<lowercase-key-with-hyphens>. So OPENAI_API_KEY โ my-agent-openai-api-key.
Container takes >4min cold-start. Heavy pip install in startup โ bake everything into the image.
PORT not listened on
App is bound to 127.0.0.1 or wrong port. Listen on 0.0.0.0:$PORT.
Permission denied on secrets
Service account needs roles/secretmanager.secretAccessor.
403 on --no-allow-unauthenticated
Caller missing roles/run.invoker.
Lost state between requests
Cloud Run is stateless. Use PostgresSaver (managed Cloud SQL) for the checkpointer; in-process state will be lost.
State on Cloud Run
Cloud Run instances are stateless and can be killed at any moment. Use a PostgresSaver checkpointer pointed at Cloud SQL.InMemorySaver will lose conversation state on every cold start. Connect via the Cloud SQL Auth Proxy or a Unix socket (/cloudsql/<project>:<region>:<instance>).
Target 3: Docker (self-hosted)
Multi-stage Dockerfile (place at server/Dockerfile)
# syntax=docker/dockerfile:1.7
FROM ghcr.io/astral-sh/uv:python3.11-bookworm-slim AS build
WORKDIR /app
COPY pyproject.toml uv.lock* ./
RUN uv sync --frozen --no-dev || uv sync --no-dev
COPY agent/ ./agent/
COPY server/ ./server/
FROM python:3.11-slim AS runtime
RUN useradd -m -u 1000 app
WORKDIR /app
COPY --from=build /app /app
ENV PATH="/app/.venv/bin:$PATH"
USER app
EXPOSE 8080
CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "8080"]
FastAPI host (server/app.py)
from __future__ import annotations
import json
from collections.abc import AsyncGenerator
from dotenv import load_dotenv
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
load_dotenv()
from agent.agent import agent # noqa: E402
app = FastAPI()
classInvokeRequest(BaseModel):
input: dict
thread_id: str | None = None@app.get("/healthz")defhealthz() -> dict:
return {"ok": True}
@app.post("/invoke")definvoke(req: InvokeRequest) -> dict:
config = {"configurable": {"thread_id": req.thread_id}} if req.thread_id else {}
return {"output": agent.invoke(req.input, config)}
@app.post("/stream")asyncdefstream(req: InvokeRequest) -> StreamingResponse:
config = {"configurable": {"thread_id": req.thread_id}} if req.thread_id else {}
asyncdefgen() -> AsyncGenerator[bytes, None]:
asyncfor chunk in agent.astream(req.input, config):
yield (json.dumps(chunk, default=str) + "\n").encode("utf-8")
return StreamingResponse(gen(), media_type="application/x-ndjson")
The thread_id parameter is what makes the deploy compatible with durable execution and HITL โ without passing it through, every request starts a fresh thread.
Build & run
# Smoke pre-flight
python evals/run.py --smoke
# Build
docker build -f server/Dockerfile -t my-agent:latest .
# Local smoke-test
docker run --rm -d --name agent-test --env-file .env -p 8080:8080 my-agent:latest
sleep 2
curl -s -X POST http://localhost:8080/invoke \
-H "Content-Type: application/json" \
-d '{"input": {"messages": [{"role": "user", "content": "hello"}]}, "thread_id": "test-1"}'
docker stop agent-test
# Run for real
docker run -d --name my-agent --env-file .env -p 8080:8080 my-agent:latest
Never bake .env into the image. Always pass --env-file at run time. The Dockerfile above does not COPY .env for this reason.
For state: point PostgresSaver at an external Postgres (RDS, Cloud SQL, etc.). Don't run Postgres inside the same container.
Scaling self-hosted
Run multiple instances behind a load balancer. With PostgresSaver, threads are correctly serialized across instances by thread_id.
Concurrency per instance: tune uvicorn --workers N --worker-class uvicorn.workers.UvicornWorker.
For very long-running threads, configure your LB's idle timeout above your max expected agent runtime.
