From 7dcaa6c7c164ea0a24e0137b72d376f202d29093 Mon Sep 17 00:00:00 2001 From: Ignazio De Santis Date: Wed, 13 May 2026 09:18:10 +0800 Subject: [PATCH 1/4] Add CI and first-agent walkthrough --- .github/workflows/ci.yml | 61 +++++++++ README.md | 53 ++++++++ app/db/connection.py | 1 - app/db/queries.py | 4 +- app/graph/graph.py | 8 +- app/graph/nodes.py | 6 +- app/graph/tools.py | 6 +- app/middleware/logging.py | 6 +- app/routers/agents.py | 9 +- app/routers/api_keys.py | 6 +- app/services/agent_service.py | 15 ++- app/services/api_key_service.py | 1 - docs/build-your-first-agent.md | 145 ++++++++++++++++++++++ pyproject.toml | 4 +- tests/test_graph/test_nodes.py | 1 - tests/test_routers/test_agents.py | 3 +- tests/test_services/test_agent_service.py | 2 +- 17 files changed, 298 insertions(+), 33 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 docs/build-your-first-agent.md diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..6410a29 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,61 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + quality: + runs-on: ubuntu-latest + + services: + postgres: + image: pgvector/pgvector:pg16 + env: + POSTGRES_USER: agent + POSTGRES_PASSWORD: localdev + POSTGRES_DB: agentdb_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U agent -d agentdb_test" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + + env: + OPENAI_API_KEY: sk-test-not-real + POSTGRES_HOST: localhost + POSTGRES_PORT: 5432 + POSTGRES_DB: agentdb_test + POSTGRES_USER: agent + POSTGRES_PASSWORD: localdev + APP_ENV: development + LOG_FORMAT: text + + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Run database migrations + run: alembic upgrade head + + - name: Run Ruff + run: ruff check app/ tests/ scripts/ + + - name: Run MyPy + run: mypy app/ --ignore-missing-imports + + - name: Run pytest + run: pytest -q --tb=short diff --git a/README.md b/README.md index 0b4f4df..64d2fe6 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,8 @@ # langgraph-fastapi-starter +[![CI](https://github.com/IgnazioDS/langgraph-fastapi-starter/actions/workflows/ci.yml/badge.svg)](https://github.com/IgnazioDS/langgraph-fastapi-starter/actions/workflows/ci.yml) +[![Release](https://img.shields.io/github/v/release/IgnazioDS/langgraph-fastapi-starter?display_name=tag)](https://github.com/IgnazioDS/langgraph-fastapi-starter/releases) + A production-grade backend template for AI agent applications. Fork this, configure four environment variables, and you have a running agent API with auth, persistence, and structured logging. Delete the example agent and build yours. @@ -7,6 +10,41 @@ and structured logging. Delete the example agent and build yours. This is not a framework. It is an opinionated starting point that exposes every decision so you can change the ones you disagree with. +## 10-Minute Path + +If you want the fastest credible evaluation path, use this sequence: + +1. Clone the repo and bring up Postgres with `make up` +2. Run migrations and create an API key +3. Change the graph in `app/graph/` +4. Send one real request to `/v1/agent/run` +5. Run lint, type-checking, and tests + +The full walkthrough lives in [docs/build-your-first-agent.md](./docs/build-your-first-agent.md). + +--- + +## Architecture + +```mermaid +flowchart LR + Client["API client or UI"] --> Router["FastAPI routers"] + Router --> Auth["API key middleware"] + Auth --> Service["Agent service"] + Service --> Graph["LangGraph runtime"] + Graph --> Retrieve["retrieve_context node"] + Retrieve --> Model["call_model node"] + Model -->|tool calls| Tools["ToolNode(TOOLS)"] + Tools --> Model + Model --> Persistence["Session + message persistence"] + Persistence --> Postgres["PostgreSQL + pgvector"] + Retrieve --> Postgres +``` + +The request path is intentionally simple: FastAPI handles transport, middleware enforces +auth, the service layer invokes a small LangGraph loop, and PostgreSQL carries both +application data and vector-backed retrieval. + --- ## What It Ships With @@ -63,6 +101,21 @@ curl -X POST http://localhost:8000/v1/agent/run \ --- +## Build Your Own Agent + +You do not need to understand the whole repository before changing the agent. In most +cases, the first working customization fits inside four files: + +- `app/graph/state.py` +- `app/graph/nodes.py` +- `app/graph/tools.py` +- `app/graph/graph.py` + +For a concrete clone -> customize -> run -> test walkthrough, see +[docs/build-your-first-agent.md](./docs/build-your-first-agent.md). + +--- + ## What to Change First The starter ships with a research assistant agent. You will replace it. diff --git a/app/db/connection.py b/app/db/connection.py index 4e803ba..bbad7b2 100644 --- a/app/db/connection.py +++ b/app/db/connection.py @@ -1,7 +1,6 @@ from collections.abc import Generator from contextlib import contextmanager -import psycopg2 from psycopg2.extensions import connection from psycopg2.pool import ThreadedConnectionPool diff --git a/app/db/queries.py b/app/db/queries.py index 77735ca..c4dc841 100644 --- a/app/db/queries.py +++ b/app/db/queries.py @@ -41,7 +41,9 @@ # ─── Sessions ───────────────────────────────────────────────────────────────── CREATE_SESSION = """ - INSERT INTO agent_sessions (id, session_id, tenant_id, created_at, last_active_at, message_count) + INSERT INTO agent_sessions ( + id, session_id, tenant_id, created_at, last_active_at, message_count + ) VALUES (%(id)s, %(session_id)s, %(tenant_id)s, NOW(), NOW(), 0) ON CONFLICT (session_id, tenant_id) DO UPDATE SET last_active_at = NOW() RETURNING id, session_id, created_at, last_active_at, message_count diff --git a/app/graph/graph.py b/app/graph/graph.py index 9939f67..677408c 100644 --- a/app/graph/graph.py +++ b/app/graph/graph.py @@ -16,13 +16,13 @@ from app.graph.tools import TOOLS -def build_graph() -> CompiledStateGraph: +def build_graph() -> CompiledStateGraph[AgentState, None, AgentState, AgentState]: """Build and compile the agent graph. Graph flow: retrieve (fetch context) → agent (LLM call) → [tools? → agent] → END """ - builder: StateGraph = StateGraph(AgentState) + builder: StateGraph[AgentState, None, AgentState, AgentState] = StateGraph(AgentState) builder.add_node("retrieve", retrieve_context) builder.add_node("agent", call_model) @@ -41,10 +41,10 @@ def build_graph() -> CompiledStateGraph: # Module-level compiled graph — initialized once at startup via init_graph(). -_graph: CompiledStateGraph | None = None +_graph: CompiledStateGraph[AgentState, None, AgentState, AgentState] | None = None -def get_graph() -> CompiledStateGraph: +def get_graph() -> CompiledStateGraph[AgentState, None, AgentState, AgentState]: """Return the compiled graph. Raises if init_graph() has not been called.""" if _graph is None: raise RuntimeError("Graph not initialized. Call init_graph() at startup.") diff --git a/app/graph/nodes.py b/app/graph/nodes.py index ffbdf97..fba60f2 100644 --- a/app/graph/nodes.py +++ b/app/graph/nodes.py @@ -10,7 +10,7 @@ import logging from typing import Literal -from langchain_core.messages import SystemMessage +from langchain_core.messages import BaseMessage, SystemMessage from langchain_openai import ChatOpenAI from app.config import config @@ -20,7 +20,7 @@ logger = logging.getLogger(__name__) -def retrieve_context(state: AgentState) -> dict: +def retrieve_context(state: AgentState) -> dict[str, list[str]]: """Retrieve relevant document chunks based on the latest user message. Runs before the model call so retrieved context can be injected as a system message. @@ -55,7 +55,7 @@ def retrieve_context(state: AgentState) -> dict: return {"context": [result]} -def call_model(state: AgentState) -> dict: +def call_model(state: AgentState) -> dict[str, list[BaseMessage] | int]: """Invoke the LLM with the current message history and any retrieved context. Binds all tools to the model so it can emit tool calls. diff --git a/app/graph/tools.py b/app/graph/tools.py index 116acda..87afb3f 100644 --- a/app/graph/tools.py +++ b/app/graph/tools.py @@ -11,7 +11,7 @@ import logging -from langchain_core.tools import tool +from langchain_core.tools import BaseTool, tool from app.config import config @@ -28,7 +28,7 @@ def web_search(query: str) -> str: ) try: - from tavily import TavilyClient # type: ignore[import-untyped] + from tavily import TavilyClient client = TavilyClient(api_key=config.tavily_api_key) results = client.search(query=query, max_results=3) @@ -100,4 +100,4 @@ def retrieve_documents(query: str, top_k: int = 5) -> str: # All tools registered with the agent. Import this list in graph.py. -TOOLS: list = [web_search, retrieve_documents] +TOOLS: list[BaseTool] = [web_search, retrieve_documents] diff --git a/app/middleware/logging.py b/app/middleware/logging.py index 34f75f6..44db8bf 100644 --- a/app/middleware/logging.py +++ b/app/middleware/logging.py @@ -2,7 +2,7 @@ import logging import time import uuid -from datetime import datetime, timezone +from datetime import UTC, datetime from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint @@ -40,8 +40,8 @@ class _JsonFormatter(logging.Formatter): """ def format(self, record: logging.LogRecord) -> str: - payload: dict = { - "timestamp": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(), + payload: dict[str, object] = { + "timestamp": datetime.fromtimestamp(record.created, tz=UTC).isoformat(), "level": record.levelname, "logger": record.name, "event": record.getMessage(), diff --git a/app/routers/agents.py b/app/routers/agents.py index 5115b67..b5957de 100644 --- a/app/routers/agents.py +++ b/app/routers/agents.py @@ -5,6 +5,7 @@ # To support multiple graphs, pass a graph_name param and select in AgentService. import logging +from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Request @@ -26,7 +27,7 @@ def get_agent_service() -> AgentService: async def run_agent( request: RunAgentRequest, http_request: Request, - agent_service: AgentService = Depends(get_agent_service), + agent_service: Annotated[AgentService, Depends(get_agent_service)], ) -> RunAgentResponse: """Run the agent for a single conversational turn. @@ -53,14 +54,14 @@ async def run_agent( message="Agent execution failed.", request_id=request_id, ).model_dump(), - ) + ) from e @router.get("/v1/agent/sessions/{session_id}", response_model=SessionResponse) async def get_session( session_id: str, http_request: Request, - agent_service: AgentService = Depends(get_agent_service), + agent_service: Annotated[AgentService, Depends(get_agent_service)], ) -> SessionResponse: """Retrieve a session's message history.""" tenant_id: str = http_request.state.tenant_id @@ -91,4 +92,4 @@ async def get_session( message="Failed to retrieve session.", request_id=request_id, ).model_dump(), - ) + ) from e diff --git a/app/routers/api_keys.py b/app/routers/api_keys.py index 15eea71..5447457 100644 --- a/app/routers/api_keys.py +++ b/app/routers/api_keys.py @@ -45,7 +45,7 @@ async def create_api_key( message="Failed to create API key.", request_id=request_id, ).model_dump(), - ) + ) from e @router.delete("/v1/keys/{key_id}", status_code=204) @@ -78,7 +78,7 @@ async def revoke_api_key( message="Failed to revoke API key.", request_id=request_id, ).model_dump(), - ) + ) from e @router.get("/v1/keys", response_model=list[ApiKeyResponse]) @@ -99,4 +99,4 @@ async def list_api_keys( message="Failed to list API keys.", request_id=request_id, ).model_dump(), - ) + ) from e diff --git a/app/services/agent_service.py b/app/services/agent_service.py index 4957a20..e8e6653 100644 --- a/app/services/agent_service.py +++ b/app/services/agent_service.py @@ -1,7 +1,8 @@ import asyncio import json import uuid -from datetime import datetime, timezone +from datetime import UTC, datetime +from typing import cast from langchain_core.messages import AIMessage, BaseMessage, HumanMessage from langgraph.graph.state import CompiledStateGraph @@ -19,7 +20,10 @@ class AgentService: - def __init__(self, graph: CompiledStateGraph) -> None: + def __init__( + self, + graph: CompiledStateGraph[AgentState, None, AgentState, AgentState], + ) -> None: self._graph = graph async def run( @@ -64,7 +68,7 @@ async def run( history.append(AIMessage(content=content)) # 3. Build initial state and invoke graph - run_id = f"run_{datetime.now(tz=timezone.utc).strftime('%Y%m%dT%H%M%S%f')}" + run_id = f"run_{datetime.now(tz=UTC).strftime('%Y%m%dT%H%M%S%f')}" initial_state: AgentState = { "messages": [*history, HumanMessage(content=message)], "session_id": session_id, @@ -76,8 +80,9 @@ async def run( } loop = asyncio.get_event_loop() - result: AgentState = await loop.run_in_executor( - None, self._graph.invoke, initial_state + result = cast( + AgentState, + await loop.run_in_executor(None, lambda: self._graph.invoke(initial_state)), ) # 4. Extract final assistant response diff --git a/app/services/api_key_service.py b/app/services/api_key_service.py index cd8c55f..afc7e75 100644 --- a/app/services/api_key_service.py +++ b/app/services/api_key_service.py @@ -2,7 +2,6 @@ import secrets import uuid from dataclasses import dataclass -from datetime import datetime import bcrypt diff --git a/docs/build-your-first-agent.md b/docs/build-your-first-agent.md new file mode 100644 index 0000000..ef4e38a --- /dev/null +++ b/docs/build-your-first-agent.md @@ -0,0 +1,145 @@ +# Build Your First Agent From This Starter + +This walkthrough is the shortest credible path from clone to a customized agent API. It assumes you want to keep the infrastructure layer intact and only replace the example research assistant with your own behavior. + +## 1. Clone and boot the starter + +```bash +git clone https://github.com/IgnazioDS/langgraph-fastapi-starter.git +cd langgraph-fastapi-starter +cp .env.example .env + +# Required for local boot +# - set OPENAI_API_KEY +# - set POSTGRES_PASSWORD + +make up +make install +make migrate +make create-key NAME="local-dev" ROLE="admin" +make dev +``` + +At this point you have: + +- FastAPI running on `http://localhost:8000` +- PostgreSQL + pgvector running in Docker +- Alembic-applied schema +- a real API key for protected routes + +## 2. Replace the example agent contract + +The starter keeps the graph surface area small on purpose. Most customizations happen in four files: + +- `app/graph/state.py` +- `app/graph/nodes.py` +- `app/graph/tools.py` +- `app/graph/graph.py` + +If you are building a support copilot, triage bot, internal assistant, or domain-specific retrieval agent, start here. + +### `app/graph/state.py` + +Add the fields your agent actually needs. The example state tracks: + +- `messages` +- `session_id` +- `tenant_id` +- `context` +- `run_id` +- `input_tokens` +- `output_tokens` + +For a support agent you might add fields like `customer_id`, `ticket_id`, or `intent`. + +### `app/graph/tools.py` + +Delete the tools you do not want and add your own. The shipped example includes: + +- `web_search` +- `retrieve_documents` + +For a real product you usually replace those with tools such as: + +- `lookup_customer_profile` +- `fetch_recent_tickets` +- `query_internal_docs` +- `create_handoff_note` + +Keep the `TOOLS` list current. That is what gets bound to the model. + +### `app/graph/nodes.py` + +The example graph has two core steps: + +- `retrieve_context`: gather retrieval context before the model call +- `call_model`: invoke the LLM with tools bound + +Typical edits: + +- change how context is assembled +- inject a domain-specific system prompt +- enforce response structure +- add validation or routing logic before the model step + +### `app/graph/graph.py` + +The default flow is: + +```text +retrieve -> agent -> tools? -> agent -> end +``` + +Keep that shape if your agent is still a standard tool-calling loop. Change it only when the product behavior genuinely needs more branches or stages. + +## 3. Run a real request against your customized graph + +Once you have edited the graph files, restart the dev server and send a request: + +```bash +curl -X POST http://localhost:8000/v1/agent/run \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{ + "session_id": "demo-support-1", + "message": "Summarize the last customer issue and suggest the next action.", + "stream": false + }' +``` + +What you should validate immediately: + +- the route authenticates correctly +- the graph executes end-to-end +- your tool wiring is reachable +- the response shape still matches the API contract + +## 4. Run the verification loop + +Use the built-in quality gates before you trust the customization: + +```bash +make lint +make typecheck +make test +``` + +If you changed graph behavior, pay special attention to: + +- `tests/test_graph/` +- `tests/test_services/` +- `tests/test_routers/test_agents.py` + +Add or update tests that reflect the new state shape, tool behavior, and expected agent responses. + +## 5. What you usually do next + +After the first successful run, most teams extend the starter in this order: + +1. Replace the example tools with product-specific integrations. +2. Add a domain-specific system prompt and response contract. +3. Add one migration for your first real data table. +4. Add tests for the new graph path before adding more branches. +5. Decide whether you need streaming, multi-tenant auth, or background jobs. + +If you can customize those four graph files, run a request, and pass the test suite, you are no longer evaluating the starter. You are building on top of it. diff --git a/pyproject.toml b/pyproject.toml index 52eac31..0bdec43 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [build-system] -requires = ["setuptools>=68"] -build-backend = "setuptools.backends.legacy:build" +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" [project] name = "langgraph-fastapi-starter" diff --git a/tests/test_graph/test_nodes.py b/tests/test_graph/test_nodes.py index 7109afc..65d0c21 100644 --- a/tests/test_graph/test_nodes.py +++ b/tests/test_graph/test_nodes.py @@ -1,4 +1,3 @@ -import pytest from langchain_core.messages import AIMessage, HumanMessage from app.graph.nodes import should_continue diff --git a/tests/test_routers/test_agents.py b/tests/test_routers/test_agents.py index a837a1c..ba6a258 100644 --- a/tests/test_routers/test_agents.py +++ b/tests/test_routers/test_agents.py @@ -1,6 +1,7 @@ +from unittest.mock import MagicMock + import pytest from httpx import AsyncClient -from unittest.mock import MagicMock @pytest.mark.asyncio diff --git a/tests/test_services/test_agent_service.py b/tests/test_services/test_agent_service.py index 814e6c5..624b208 100644 --- a/tests/test_services/test_agent_service.py +++ b/tests/test_services/test_agent_service.py @@ -1,4 +1,4 @@ -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import MagicMock, patch import pytest from langchain_core.messages import AIMessage From 454c60fb5ea98e19c1db8387326c0b5e88415bf6 Mon Sep 17 00:00:00 2001 From: Ignazio De Santis Date: Wed, 13 May 2026 09:21:54 +0800 Subject: [PATCH 2/4] Fix CI database URL --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6410a29..64d7f48 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,6 +26,7 @@ jobs: env: OPENAI_API_KEY: sk-test-not-real + DATABASE_URL: postgresql://agent:localdev@localhost:5432/agentdb_test POSTGRES_HOST: localhost POSTGRES_PORT: 5432 POSTGRES_DB: agentdb_test From 4e28fb8e1f4b2b381555a2fac78d02cdba97a5f8 Mon Sep 17 00:00:00 2001 From: Ignazio De Santis Date: Wed, 13 May 2026 09:23:59 +0800 Subject: [PATCH 3/4] Initialize test app resources --- tests/conftest.py | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index bb8b591..3800d1d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -19,6 +19,10 @@ if TEST_DATABASE_URL: os.environ["DATABASE_URL"] = TEST_DATABASE_URL +# Ensure required settings exist before test modules import app code. +os.environ.setdefault("OPENAI_API_KEY", "sk-test-not-real") +os.environ.setdefault("POSTGRES_PASSWORD", "localdev") + @pytest.fixture(scope="session") def _patch_config() -> Generator[None, None, None]: @@ -32,7 +36,19 @@ def _patch_config() -> Generator[None, None, None]: @pytest.fixture(scope="session") -async def client(_patch_config: None) -> AsyncGenerator[AsyncClient, None]: +def _app_runtime(_patch_config: None) -> Generator[None, None, None]: + """Initialize shared app resources needed by tests.""" + from app.db.connection import close_pool, init_pool + from app.graph.graph import init_graph + + init_pool() + init_graph() + yield + close_pool() + + +@pytest.fixture(scope="session") +async def client(_app_runtime: None) -> AsyncGenerator[AsyncClient, None]: """Session-scoped async HTTP client wired to the FastAPI app.""" from app.main import create_app @@ -44,7 +60,7 @@ async def client(_patch_config: None) -> AsyncGenerator[AsyncClient, None]: @pytest.fixture -async def admin_key(client: AsyncClient) -> str: +async def admin_key(client: AsyncClient, _app_runtime: None) -> str: """Create a fresh admin key for a test and return the plaintext.""" # Bootstrap: we need one key to create more keys. # For the first key, call the service directly (bypasses auth middleware). From 5be113a36d7195beef10e706bea7bbc3a2ac2680 Mon Sep 17 00:00:00 2001 From: Ignazio De Santis Date: Wed, 13 May 2026 09:25:57 +0800 Subject: [PATCH 4/4] Normalize structured HTTP errors --- app/main.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/app/main.py b/app/main.py index 5cc786d..fbbb532 100644 --- a/app/main.py +++ b/app/main.py @@ -2,7 +2,7 @@ from collections.abc import AsyncGenerator from contextlib import asynccontextmanager -from fastapi import FastAPI, Request +from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from app.config import config @@ -53,6 +53,20 @@ def create_app() -> FastAPI: app.include_router(agents_router) app.include_router(keys_router) + @app.exception_handler(HTTPException) + async def http_exception_handler(request: Request, exc: HTTPException) -> JSONResponse: + if isinstance(exc.detail, dict): + return JSONResponse(status_code=exc.status_code, content=exc.detail) + + return JSONResponse( + status_code=exc.status_code, + content=ErrorResponse( + code="HTTP_ERROR", + message=str(exc.detail), + request_id=getattr(request.state, "request_id", None), + ).model_dump(), + ) + @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse: logger.error(