Module 15 · Deployment

Lab 15 · Ship it

Turn the Northwind support agent into a service other people can call, twice: on the local Agent Server with langgraph dev, and on a licence-free FastAPI wrapper you own.

60-75 minAdvancedAzure OpenAIlanggraph devlanggraph.jsonAgent Serverlanggraph_sdkFastAPISSEAsyncSqliteSaver

What you will build

You take the Northwind support agent from a graph that runs on my laptop to a service other people can call, two ways:

Agent Server. You package the agent as an app/ folder plus langgraph.json, serve it with langgraph dev, and drive the server's data model from the Python SDK: assistant, thread, streaming run, background run and cron.
Your own FastAPI wrapper. You serve the same graph from about eighty lines of FastAPI, with API-key auth, Server-Sent Events streaming and SQLite persistence, and prove that a conversation survives a restart.

Doing both is the point: at the end you can say which one you would choose for a given team, and why.

What you will learn

How to package a graph for deployment with a no-argument factory and langgraph.json
How the Agent Server models work: assistants, threads, runs and crons
How to stream, queue and join runs with the langgraph_sdk client
How to mount custom routes into the Agent Server
What you take over when you write your own wrapper: persistence, auth and streaming

You use two terminals in this lab: one runs a server, the other runs the client. Activate the virtual environment in both.

Check your prerequisites

Done

You need three things on your machine and three values from your instructor.

Python 3.11, 3.12 or 3.13. Check with the command below. On Windows, install it from python.org and tick Add python.exe to PATH.
A code editor. VS Code with the Python extension works well.
A terminal. PowerShell on Windows, Terminal on macOS or Linux.

PowerShell

python --version

From your instructor you get an Azure OpenAI endpoint, an API key and a deployment name. Keep the key private: never paste it into code or share it in chat.

Create the project folder and virtual environment

Done

Create a new folder for this lab, open it in a terminal, and create a virtual environment. A virtual environment is a private copy of Python for this project, so the libraries you install here never clash with other projects.

PowerShell

mkdir C:\labs\lab-15-ship-it
cd C:\labs\lab-15-ship-it
python -m venv .venv
.venv\Scripts\Activate.ps1

✓

Keep lab folders on a short path such as C:\labs. Deep folders, for example inside OneDrive or Desktop, can exceed the Windows path limit. Some libraries then fail to load with The filename or extension is too long. OneDrive syncing can also lock files the labs write.

When activation works, your prompt starts with (.venv). Activate it again every time you open a new terminal for this lab. Open the folder in VS Code with code . if you like.

PowerShell says running scripts is disabled on this system? Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned, answer Y, and activate again.

This is what your folder will look like at the end of the lab:

lab-15-ship-it/ ├── app/ │ ├── __init__.py │ ├── agent.py │ └── server.py ├── .env ├── .langgraph_api ├── chat_client.py ├── langgraph.json ├── model_setup.py ├── northwind_threads.db ├── requirements.txt ├── ship_it.py └── wrapper.py

Files marked as generated are created when you run the code. You do not create them yourself.

Install the libraries

Done

Create a file named requirements.txt in your lab folder with the content below. The versions are pinned to the ones this lab was tested with.

requirements.txt

langchain==1.4.2
langchain-core==1.6.4
langchain-openai==1.6.2
python-dotenv==1.2.3
langgraph==1.2.11
langgraph-cli[inmem]==0.4.31
langgraph-api==0.14.0
langgraph-sdk==0.4.4
langgraph-checkpoint-sqlite==3.1.1
aiosqlite==0.22.1
fastapi==0.141.1
uvicorn==0.52.4
httpx==0.28.1
colorama==0.4.6

langgraph-cli[inmem] gives you the langgraph command and the in-memory Agent Server (langgraph-api) behind langgraph dev. The dev server needs no licence key and no LangSmith account.
langgraph-sdk is the Python client for the server.
fastapi, uvicorn, langgraph-checkpoint-sqlite and aiosqlite run your own wrapper, and httpx is the HTTP client for the test script.
colorama is needed on Windows: the server's coloured console log crashes at startup without it (ConsoleRenderer with colors=True on Windows requires the colorama package).

This install is larger than in earlier labs and can take a few minutes.

Then install everything into your virtual environment. It takes a minute or two.

PowerShell

python -m pip install --upgrade pip
pip install -r requirements.txt

Add your Azure OpenAI settings

Done

Create a file named .env (the name starts with a dot) in your lab folder. Paste the template and replace the three placeholders with the values from your instructor.

.env

# Azure OpenAI settings. Your instructor gives you these three values.
AZURE_OPENAI_ENDPOINT=https://<your-resource>.openai.azure.com/openai/v1
AZURE_OPENAI_API_KEY=<paste-your-key-here>
AZURE_OPENAI_DEPLOYMENT=<your-chat-deployment-name>

# Only used when your endpoint does NOT end in /openai/v1.
OPENAI_API_VERSION=2025-04-01-preview

AZURE_OPENAI_ENDPOINT is your resource URL. If it ends in /openai/v1, keep it that way. If it is just https://<resource>.openai.azure.com/, paste that; the helper in the next step handles both.
AZURE_OPENAI_API_KEY is your secret key.
AZURE_OPENAI_DEPLOYMENT is the deployment name, for example gpt-5-mini. On Azure you call a deployment, not a model name.

Never commit .env to Git or share it. If you use Git, add a .gitignore file containing .env and .venv/.

Create the Azure OpenAI helper

Done

Every lab talks to Azure OpenAI through one small helper file. Create model_setup.py and add the blocks below one after another. You write it once per lab folder; the lab code then only needs from model_setup import get_model.

Block 1Imports, .env loading and a console fix

load_dotenv() reads your .env file and puts every line into environment variables, so your key never appears in code.

The loop below it switches the console to UTF-8. Windows terminals default to an older encoding and crash when a model prints curly quotes or dashes.

model_setup.py (add to the end of the file)

"""Azure OpenAI model setup shared by every lab.

Import it in lab code as:  from model_setup import get_model, get_embeddings
"""

import asyncio
import os
import sys

from dotenv import load_dotenv
from langchain_core.embeddings import Embeddings
from langchain_core.language_models import BaseChatModel

# Read AZURE_OPENAI_* values from the .env file next to your lab code.
load_dotenv()

# Windows consoles default to cp1252 and crash on the curly quotes models love to print.
for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, "reconfigure"):
        _stream.reconfigure(encoding="utf-8", errors="replace")

Block 2Read settings safely

_setting() fetches a value from the environment and stops with a friendly message if it is missing, instead of a long stack trace.

_uses_v1_api() checks the shape of your endpoint. Azure offers two:

https://<resource>.openai.azure.com/openai/v1 is the newer OpenAI-compatible API.
https://<resource>.openai.azure.com/ is the classic API that needs an api_version.

model_setup.py (add to the end of the file)

def _setting(name: str) -> str:
    """Return a required setting from .env, or stop with a readable message."""
    value = os.getenv(name, "").strip()
    if not value:
        sys.exit(f"Missing {name} in your .env file. Open .env and fill it in.")
    return value


def _uses_v1_api(endpoint: str) -> bool:
    """True for the OpenAI-compatible Azure endpoint that ends in /openai/v1."""
    return endpoint.rstrip("/").endswith("/openai/v1")

Block 3Build the chat model

get_model() returns a LangChain chat model connected to your deployment.

With a /openai/v1 endpoint it uses ChatOpenAI with base_url. On Azure the deployment name goes in model.
With a classic endpoint it uses AzureChatOpenAI with azure_deployment and api_version.

Either way the lab code just calls get_model() and never cares which one it got.

model_setup.py (add to the end of the file)

def get_model(deployment: str | None = None, **kwargs) -> BaseChatModel:
    """Return a chat model for your Azure OpenAI deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import ChatOpenAI

        return ChatOpenAI(model=name, base_url=endpoint, api_key=api_key, **kwargs)

    from langchain_openai import AzureChatOpenAI

    return AzureChatOpenAI(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
        **kwargs,
    )

Block 4Build the embeddings model

get_embeddings() does the same for embeddings, which turn text into vectors for search. Only the retrieval labs use it, but keeping one helper for every lab is simpler.

model_setup.py (add to the end of the file)

def get_embeddings(deployment: str | None = None) -> Embeddings:
    """Return an embeddings model for your Azure OpenAI embeddings deployment."""
    endpoint = _setting("AZURE_OPENAI_ENDPOINT")
    api_key = _setting("AZURE_OPENAI_API_KEY")
    name = deployment or _setting("AZURE_OPENAI_EMBEDDING_DEPLOYMENT")

    if _uses_v1_api(endpoint):
        from langchain_openai import OpenAIEmbeddings

        return OpenAIEmbeddings(model=name, base_url=endpoint, api_key=api_key)

    from langchain_openai import AzureOpenAIEmbeddings

    return AzureOpenAIEmbeddings(
        azure_deployment=name,
        azure_endpoint=endpoint,
        api_key=api_key,
        api_version=os.getenv("OPENAI_API_VERSION", "2025-04-01-preview"),
    )

Block 5A Windows fix for async code

use_stable_event_loop() matters only for the async MCP lab. The default Windows event loop can crash on exit after async network sessions close. On macOS and Linux it does nothing.

model_setup.py (add to the end of the file)

def use_stable_event_loop() -> None:
    """On Windows, use the Selector event loop so async MCP code exits cleanly."""
    if sys.platform == "win32":
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

Check that the helper can reach your deployment with a one-line test:

PowerShell

python -c "from model_setup import get_model; print(get_model().invoke('Say hello in 5 words').text)"

You should see a short greeting. If you see an error instead, check the troubleshooting step at the end.

Create the app package

Done

A deployable agent is a normal Python package. Create a folder named app inside your lab folder, and in it an almost empty __init__.py. It marks app as a package, so wrapper.py can later write from app.agent import build_agent.

app/__init__.py

"""The deployable Northwind package: the graph and its custom routes."""

Write the deployable agent

Done

This is the Northwind support agent with three tools. The important part is how it is built: through a factory function the server can call.

Create an empty file named app/agent.py in your lab folder. Then copy each block below into it, in order, one after another.

Block 1Imports

CompiledStateGraph is the type create_agent returns; it is only used for type hints. get_model comes from your helper in the lab folder, one level above app/. The server can import it because langgraph.json lists the lab folder as a dependency (you add that in a moment).

app/agent.py (add to the end of the file)

"""The Northwind support agent, packaged for deployment.

langgraph.json points at make_graph below. The FastAPI wrapper (wrapper.py) uses build_agent
with its own checkpointer.
"""

from typing import Any

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.graph.state import CompiledStateGraph

from model_setup import get_model

Block 2Data and prompt

Two dictionaries stand in for the order and customer systems. Each order names its customer, so a follow-up question like which customer placed it? can be answered from the conversation history.

app/agent.py (add to the end of the file)

ORDERS = {
    "A-100": "shipped via DHL, arriving 2026-09-15, customer C-1",
    "A-101": "processing, expected 2026-09-18, customer C-2",
    "A-102": "delivered 2026-09-09, customer C-1",
}

CUSTOMERS = {
    "C-1": "Acme GmbH, enterprise plan, EU",
    "C-2": "Bluefin Ltd, starter plan, US",
}

SYSTEM_PROMPT = (
    "You are Northwind Cloud's support agent. Answer order and account questions with the "
    "tools, quote the order id in every answer, and keep replies under four sentences."
)

Block 3Three tools

get_order_status and lookup_customer read the dictionaries. open_ticket pretends to open a support ticket; its docstring says to use it only when the customer asks to escalate.

app/agent.py (add to the end of the file)

@tool
def get_order_status(order_id: str) -> str:
    """Look up the shipping status of a Northwind order, such as A-100."""
    return f"Order {order_id}: {ORDERS.get(order_id.upper(), 'not found')}."


@tool
def lookup_customer(customer_id: str) -> str:
    """Look up a Northwind customer's name, plan tier and region, such as C-1."""
    return f"Customer {customer_id}: {CUSTOMERS.get(customer_id.upper(), 'not found')}."


@tool
def open_ticket(order_id: str, summary: str) -> str:
    """Open a support ticket for an order. Use only when the customer asks to escalate."""
    return f"Ticket NW-4711 opened for {order_id}: {summary}"

Block 4The builder and the factory

Two functions, on purpose:

build_agent(model, checkpointer) accepts the things a caller may want to inject. The FastAPI wrapper passes its own SQLite checkpointer; a test could pass a fake model.
make_graph() is what the Agent Server calls. A deployed factory takes no arguments and passes no checkpointer, because the server supplies persistence itself.

Nothing in this file builds a model or opens a connection at import time. The server imports the module in every worker, so import-time work would slow every start.

app/agent.py (add to the end of the file)

def build_agent(model: Any = None, checkpointer: Any = None) -> CompiledStateGraph:
    """Build the agent. The wrapper passes a checkpointer; tests can pass a fake model."""
    return create_agent(
        model=model or get_model(),
        tools=[get_order_status, lookup_customer, open_ticket],
        system_prompt=SYSTEM_PROMPT,
        checkpointer=checkpointer,
        name="northwind-support",
    )


def make_graph() -> CompiledStateGraph:
    """The factory langgraph.json points at.

    It takes no arguments and passes no checkpointer: the Agent Server supplies persistence.
    """
    return build_agent()

Add custom routes (optional)

Done

The Agent Server can mount your own FastAPI app next to its built-in routes. Use it for things that are not agent logic: health details, a version endpoint, a feedback proxy. This step is optional but short; if you skip it, also remove the http section from langgraph.json in the next step.

Create an empty file named app/server.py in your lab folder. Then copy each block below into it, in order, one after another.

Block 1Imports

A plain FastAPI app. BOOT_INFO holds a little startup information that the routes report.

app/server.py (add to the end of the file)

"""Custom HTTP routes mounted into the Agent Server through langgraph.json ("http": {"app": ...})."""

import os
import time
import uuid
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel

BOOT_INFO: dict[str, str] = {}

Block 2Lifespan

The lifespan context manager runs once per worker: code before yield on startup, code after it on shutdown. Here it records the revision and environment, read from environment variables a deployment would set.

app/server.py (add to the end of the file)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Run once per worker process: warm-up on the way in, cleanup on the way out."""
    BOOT_INFO["revision"] = os.environ.get("NORTHWIND_REVISION", "dev")
    BOOT_INFO["environment"] = os.environ.get("NORTHWIND_ENV", "local")
    yield
    BOOT_INFO.clear()


app = FastAPI(lifespan=lifespan)

Block 3Middleware for every response

This middleware wraps every request, including the server's own /threads and /runs routes. It adds an x-request-id header (reusing the caller's if sent) and a server-timing header, which helps correlate logs during an incident.

app/server.py (add to the end of the file)

@app.middleware("http")
async def request_context(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    """Give every response, the server's own routes included, a correlation id and a timing."""
    request_id = request.headers.get("x-request-id", str(uuid.uuid4()))
    started = time.perf_counter()
    response = await call_next(request)
    response.headers["x-request-id"] = request_id
    response.headers["server-timing"] = f"app;dur={(time.perf_counter() - started) * 1000:.1f}"
    return response

Block 4The routes

/northwind/info answers which revision is serving traffic?, the first question in a rollout.
/northwind/ready reports readiness of your dependencies; here it checks that a deployment name is configured.
/northwind/feedback is a stub for the classic reason to add a route: proxying feedback so the browser never holds a LangSmith key.

Keep agent logic in the graph. Code here is invisible to threads, checkpoints and traces.

app/server.py (add to the end of the file)

@app.get("/northwind/info")
async def info() -> dict[str, str]:
    """Which revision is actually serving traffic: the first question in a rollout."""
    return {"service": "northwind-support", **BOOT_INFO}


@app.get("/northwind/ready")
async def ready() -> JSONResponse:
    """Readiness for *our* dependencies. GET /ok already covers the server itself."""
    healthy = bool(BOOT_INFO) and bool(os.environ.get("AZURE_OPENAI_DEPLOYMENT"))
    return JSONResponse({"ready": healthy}, status_code=200 if healthy else 503)


class Feedback(BaseModel):
    """A thumbs up or down from the UI."""

    run_id: str
    score: int


@app.post("/northwind/feedback")
async def feedback(body: Feedback) -> dict[str, str | int]:
    """Proxy feedback so the browser never holds a LangSmith key (a stub in this lab)."""
    return {"recorded": body.run_id, "score": body.score}

Write langgraph.json

Done

langgraph.json tells the langgraph CLI what to serve. Create it in the lab folder (not inside app/).

langgraph.json

{
  "python_version": "3.11",
  "dependencies": ["."],
  "graphs": {
    "support": "./app/agent.py:make_graph"
  },
  "env": ".env",
  "http": {
    "app": "./app/server.py:app"
  }
}

dependencies: ["."] adds the lab folder to the import path, which is how app/agent.py finds model_setup.py. In a Docker build it is also what gets installed.
graphs maps a graph id (support) to file:function. Pointing at a factory rather than a module-level graph keeps imports cheap.
env loads your .env into the server process.
http.app mounts the custom routes from app/server.py.

You check this file with langgraph validate right before you start the server.

Write the SDK tour

Done

The Agent Server exposes a small data model: an assistant is a named, versioned configuration of a graph; a thread is durable conversation state; a run is one piece of work on a thread; a cron is a run on a schedule. ship_it.py walks through all four with the synchronous SDK client.

Create an empty file named ship_it.py in your lab folder. Then copy each block below into it, in order, one after another.

Block 1Imports and settings

get_sync_client returns a client for a running server. Its async twin, get_client, has the same methods. langgraph dev listens on port 2024 by default.