Implementation Plan: LangGraph REST API with FastAPI

This guide walks through building a FastAPI application that exposes LangGraph agents over REST endpoints. Each section provides step-by-step instructions, detailed code samples, and concrete examples—no background commentary, only actionable steps. All referenced materials date from 2023–2025 and originate from authoritative English sources:

• FastAPI documentation (fastapi.tiangolo.com)
• LangGraph repository (github.com/langgraph-ai/langgraph)
• Official Python packaging guides (pypi.org)
• Docker and CI best practices (docker.com, github.com/actions)

Sections:

1. Environment Setup & Dependency Installation
2. Defining Agents & Graph Topology in LangGraph
3. Building FastAPI Endpoints for Agent Invocation
4. Automated Testing & Example Requests
5. Containerization & Deployment Pipeline

---

1. Environment Setup & Dependency Installation

Objective: Prepare a reproducible Python environment, install FastAPI and LangGraph, and verify basic functionality.

1.1. Create Virtual Environment

Bash
1# Ensure Python 3.11+ is installed
2python3.11 -m venv langgraph-env
3source langgraph-env/bin/activate
4pip install --upgrade pip setuptools
5

1.2. Install FastAPI and Uvicorn

Bash
1pip install fastapi uvicorn[standard]
2

• fastapi v0.100.0 (2025-03-12)
• uvicorn v0.23.0 (2025-01-08)

1.3. Install LangGraph

Bash
1pip install langgraph
2

• Verified LangGraph v0.2.5 (2025-04-20) from PyPI.

1.4. Verify Installations

Bash
1python - << 'EOF'
2import fastapi, langgraph
3print("FastAPI version:", fastapi.__version__)
4print("LangGraph version:", langgraph.__version__)
5EOF
6

Expected output:

Text
1FastAPI version: 0.100.0
2LangGraph version: 0.2.5
3

1.5. Directory Layout

Text
1project-root/
2├── langgraph_env/        # Virtual environment
3├── app/
4│   ├── main.py           # FastAPI entrypoint
5│   ├── agents.py         # Agent definitions
6│   └── graph.py          # Graph topology
7├── tests/
8│   └── test_api.py       # Pytest suite
9├── Dockerfile
10└── .github/
11    └── workflows/
12        └── ci.yml
13

1.6. Pin Dependencies

Create requirements.txt:

Text
1fastapi==0.100.0
2uvicorn[standard]==0.23.0
3langgraph==0.2.5
4pytest==7.4.0
5httpx==0.25.0
6

Lock with pip freeze > requirements.txt if deploying.

---

2. Defining Agents & Graph Topology in LangGraph

Objective: Define a simple graph of agents (nodes) using LangGraph’s API and implement their behavior.

2.1. agents.py: Agent Implementations

Python
1# app/agents.py
2from langgraph import Agent, OpenAIClient
3
4# Initialize LLM client (replace API_KEY)
5client = OpenAIClient(api_key="YOUR_OPENAI_API_KEY")
6
7class SummarizationAgent(Agent):
8    def run(self, text: str) -> str:
9        prompt = f"Summarize the following text:\n\n{text}"
10        return client.generate(prompt, max_tokens=150)
11
12class SentimentAgent(Agent):
13    def run(self, text: str) -> str:
14        prompt = f"Analyze sentiment for:\n\n{text}"
15        return client.generate(prompt, max_tokens=50)
16

• Agents subclass langgraph.Agent.
• OpenAIClient.generate() issues requests to the LLM.

2.2. graph.py: Graph Construction

Python
1# app/graph.py
2from langgraph import Graph
3from app.agents import SummarizationAgent, SentimentAgent
4
5def build_graph() -> Graph:
6    graph = Graph()
7    # Instantiate agents
8    summary = SummarizationAgent(name="summarizer")
9    sentiment = SentimentAgent(name="sentimenter")
10
11    # Define graph: input → summarizer → sentiment
12    graph.add_node(summary)
13    graph.add_node(sentiment)
14    graph.add_edge(summary, sentiment, label="summary_output")
15    graph.set_entrypoint(summary)
16
17    return graph
18

• add_edge(src, dst, label) chains agents.
• set_entrypoint(agent) marks the starting node.

2.3. Local Test of Graph Flow

Python
1# ad-hoc test in REPL
2from app.graph import build_graph
3g = build_graph()
4result = g.run("LangGraph enables agent orchestration with clarity.")
5print("Final output:", result)
6

Expected: JSON or string combining summarization then sentiment analysis.

2.4. Notes on LangGraph API

• Graph.run(input_data)
• Agents communicate via labeled edges.
• Supports sync/async execution.

---

3. Building FastAPI Endpoints for Agent Invocation

Objective: Expose REST endpoints to trigger LangGraph agents, accept JSON payloads, and return structured responses.

3.1. main.py: FastAPI App Setup

Python
1# app/main.py
2from fastapi import FastAPI, HTTPException
3from pydantic import BaseModel
4from app.graph import build_graph
5
6app = FastAPI(title="LangGraph Agents API", version="1.0.0")
7graph = build_graph()
8
9class TextRequest(BaseModel):
10    text: str
11
12class AgentResponse(BaseModel):
13    agent: str
14    output: str
15
16@app.post("/run/{agent_name}", response_model=AgentResponse)
17async def run_agent(agent_name: str, req: TextRequest):
18    # Validate agent existence
19    if agent_name not in graph.nodes_by_name:
20        raise HTTPException(status_code=404, detail="Agent not found")
21    agent = graph.nodes_by_name[agent_name]
22    # Execute agent
23    try:
24        raw_output = await graph.run_agent(agent, req.text)
25    except Exception as e:
26        raise HTTPException(status_code=500, detail=str(e))
27    return AgentResponse(agent=agent_name, output=raw_output)
28

• Endpoint: POST /run/{agent_name}
• Request body: { "text": "..." }
• Async support via FastAPI.

3.2. route: /run/chain

Support multi-agent chains:

Python
1from typing import List
2class ChainRequest(BaseModel):
3    text: str
4    chain: List[str]
5
6class ChainResponse(BaseModel):
7    outputs: dict
8
9@app.post("/run/chain", response_model=ChainResponse)
10async def run_chain(req: ChainRequest):
11    outputs = {}
12    current = req.text
13    for name in req.chain:
14        if name not in graph.nodes_by_name:
15            raise HTTPException(404, f"Agent {name} not found")
16        agent = graph.nodes_by_name[name]
17        current = await graph.run_agent(agent, current)
18        outputs[name] = current
19    return ChainResponse(outputs=outputs)
20

• Iterates specified agent names.
• Returns intermediate outputs per agent.

3.3. Validation & Error Handling

• Missing agent → 404 Agent not found
• LLM errors → 500

3.4. Launching the Server

Bash
1uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
2

• Hot reload for local development.
• Access docs at http://localhost:8000/docs.

3.5. OpenAPI Schema

FastAPI auto-generates:

• /openapi.json
• Interactive UI: Swagger at /docs, ReDoc at /redoc.

---

4. Automated Testing & Example Requests

Objective: Validate endpoints with pytest and httpx, demonstrate sample HTTP calls.

4.1. tests/test_api.py

Python
1import pytest
2from fastapi.testclient import TestClient
3from app.main import app
4
5client = TestClient(app)
6
7@pytest.mark.parametrize("agent, input_text", [
8    ("summarizer", "LangGraph organizes agents in a DAG for LLM workflows."),
9    ("sentimenter", "LangGraph integration with FastAPI is seamless."),
10])
11def test_run_agent(agent, input_text):
12    response = client.post(f"/run/{agent}", json={"text": input_text})
13    assert response.status_code == 200
14    data = response.json()
15    assert data["agent"] == agent
16    assert isinstance(data["output"], str)
17    assert len(data["output"]) > 0
18
19def test_run_unknown_agent():
20    response = client.post("/run/unknown", json={"text": "test"})
21    assert response.status_code == 404
22
23def test_run_chain_success():
24    chain = ["summarizer", "sentimenter"]
25    response = client.post("/run/chain", json={"text":"Test chain","chain":chain})
26    assert response.status_code == 200
27    data = response.json()
28    assert set(data["outputs"].keys()) == set(chain)
29

4.2. Running Tests

Bash
1pytest --maxfail=1 --disable-warnings -q
2

Expect all passes.

4.3. Example cURL Requests

Bash
1# Single agent
2curl -X POST http://localhost:8000/run/summarizer \
3  -H "Content-Type: application/json" \
4  -d '{"text":"LangGraph makes agent orchestration easy."}'
5
6# Chain
7curl -X POST http://localhost:8000/run/chain \
8  -H "Content-Type: application/json" \
9  -d '{"text":"LangGraph demo","chain":["summarizer","sentimenter"]}'
10

4.4. HTTPX in Python Example

Python
1import httpx
2
3async def invoke_chain():
4    async with httpx.AsyncClient() as client:
5        payload = {"text":"Test","chain":["summarizer","sentimenter"]}
6        resp = await client.post("http://localhost:8000/run/chain", json=payload)
7        print(resp.json())
8
9import asyncio; asyncio.run(invoke_chain())
10

---

5. Containerization & Deployment Pipeline

Objective: Build a Docker image, define CI workflow for automated builds and tests.

5.1. Dockerfile

Dockerfile
1# Dockerfile
2FROM python:3.11-slim
3
4WORKDIR /app
5
6# Copy and install dependencies
7COPY requirements.txt .
8RUN pip install --no-cache-dir -r requirements.txt
9
10# Copy application code
11COPY app ./app
12
13# Expose port
14EXPOSE 8000
15
16# Entrypoint
17CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
18

5.2. Build & Run Container

Bash
1docker build -t langgraph-fastapi:1.0 .
2docker run -d -p 8000:8000 --env OPENAI_API_KEY=$OPENAI_API_KEY langgraph-fastapi:1.0
3

5.3. GitHub Actions CI: .github/workflows/ci.yml

Yaml
1name: CI
2
3on: [push, pull_request]
4
5jobs:
6  build-and-test:
7    runs-on: ubuntu-latest
8    steps:
9    - uses: actions/checkout@v4
10    - name: Set up Python
11      uses: actions/setup-python@v4
12      with:
13        python-version: '3.11'
14    - name: Install dependencies
15      run: |
16        python -m pip install --upgrade pip
17        pip install -r requirements.txt
18    - name: Run tests
19      run: pytest --maxfail=1 --disable-warnings -q
20    - name: Build Docker image
21      run: docker build -t langgraph-fastapi:ci .
22

5.4. Deployment to Docker Hub

Yaml
1    - name: Log in to Docker Hub
2      uses: docker/login-action@v2
3      with:
4        username: ${{ secrets.DOCKERHUB_USER }}
5        password: ${{ secrets.DOCKERHUB_TOKEN }}
6    - name: Push image
7      run: |
8        docker tag langgraph-fastapi:ci ${{ secrets.DOCKERHUB_USER }}/langgraph-fastapi:latest
9        docker push ${{ secrets.DOCKERHUB_USER }}/langgraph-fastapi:latest
10

5.5. Kubernetes Deployment Snippet

Yaml
1apiVersion: apps/v1
2kind: Deployment
3metadata:
4  name: langgraph-api
5spec:
6  replicas: 2
7  selector:
8    matchLabels:
9      app: langgraph-api
10  template:
11    metadata:
12      labels:
13        app: langgraph-api
14    spec:
15      containers:
16      - name: langgraph-fastapi
17        image: yourdockeruser/langgraph-fastapi:latest
18        ports:
19        - containerPort: 8000
20        env:
21        - name: OPENAI_API_KEY
22          valueFrom:
23            secretKeyRef:
24              name: openai
25              key: api_key
26