Starter Kit: FastAPI + LangGraph + LangChain by File

The idea is not to show 500 files, but to assemble a minimum set that is already fairly similar to production: there are ‘FastAPI’, service layer, compiled graph, persistence, streaming, ‘resume’, health endpoints, Docker and tests. After that, you can already increase the domain logic.

START COMPLEX

public HTTP API
  ↓
service layer
  ↓
LangGraph runtime
  ↓
LangChain components
  ↓
external persistence + external tools

If you make only one sync endpoint, the project will look like a demo. As soon as streaming and durable human-in-the-loop paths through ‘resume’ appear, code behaves like a real service rather than a playground.

This is one of the main engineering ideas. An HTTP process doesn’t have to be the only place where a conversation lives. It is better to keep the service as stateless as possible, and the thread state and memory are transferred to external persistence, so that you can experience restarts and scaling.

web process:
  preferably disposable and replaceable

conversation state:
  live outside the process

At the start, it is more important for you to learn how to properly separate layers: input API, config, state/context, tools, graph build, service orchestration, persistence, tests. If these boundaries are correct, then you can easily change prompts, tools and even entire graph flow without destroying the base.

In production, most of the time is eaten not by models, but by glue code and lifecycle: service start, config, resources, streaming, errors, resumes, health checks and tracing. Starter kit is valuable because it shows these boring but real things at once.

This structure is already sufficient for a live microservice and is not overloaded. There's a clear input, a separate agent layer, an infrastructure layer and tests.

REPO TREE

app/
  main.py
  core/
    settings.py
    logging.py
  api/
    deps.py
    routers/
      chat.py
      health.py
  agent/
    contracts/
      state.py
      context.py
      schemas.py
    components/
      models.py
      prompts.py
      tools.py
      middleware.py
    graph/
      nodes.py
      routes.py
      build.py
    services/
      chat_service.py
      stream_service.py
  infra/
    persistence.py
    mcp_clients.py
tests/
Dockerfile
pyproject.toml

In the starting skeleton, it is useful to immediately fix the dependencies so that the project is reproducible. This service usually requires fastapi, uvicorn, langchain, langgraph, pydantic, pydantic-settings and test stack.

dependencies = [
  "fastapi",
  "uvicorn[standard]",
  "langchain",
  "langgraph",
  "pydantic",
  "pydantic-settings",
]

dev = [
  "pytest",
  "httpx",
]

It seems trivial, but separation is very helpful. `tests/` shall not be embedded in graph modules. They look at the system from the outside and check HTTP, service wiring and graph behavior as separate layers.

If you don't immediately decide where you have 'thread id', where 'run id', as routers and service methods are called, then things get confusing. Starter kit is also useful because it sets the project language: “chat”, “stream”, “resume”, “health”, “build graph”, “RequestContext”, “AgentState”.

HTTP:
  /v1/chat
  /v1/chat/stream
  /v1/threads/{thread_id}/resume

Python:
  ChatRequest
  ChatResponse
  ChatService
  build_graph
  RequestContext
  AgentState

At the start, it is more useful to assemble one quality service with all layers than it is premature to break everything down into “api-service”, “graph-service”, “tool-service”, “mcp-service”. You will always have time to divide into several services later, when real operational pain appears.

Almost any living system quickly begins to live in different environments: local, staging, prod. Therefore, the settings of the model, DSN, timeouts and flags are better put in the settings model, rather than smeared into modules.

class Settings(BaseSettings):
    app_name: str = "starter-agent"
    model_name: str = "provider:model"
    api_prefix: str = "/v1"
    log_level: str = "INFO"
    request_timeout_s: int = 60
    database_url: str

def get_settings() -> Settings:
    return Settings()

This is one of the main production patterns of FastAPI. At the start of the application, you once create settings, persistence, compiled graph and service objects, fold them into “app.state”, and the routes then only get the finished objects.

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    checkpointer, store = await build_persistence(settings)
    graph = build_graph(settings=settings, checkpointer=checkpointer, store=store)
    app.state.chat_service = ChatService(graph=graph, settings=settings)
    yield

app = FastAPI(lifespan=lifespan)
app.include_router(chat_router)
app.include_router(health_router)

Dependencies in FastAPI is convenient to use not only for auth, but also for access to objects from "app.state". It is also good to build a request-scoped context, which then falls into the graph runtime.

def get_chat_service(request: Request) -> ChatService:
    return request.app.state.chat_service

def get_request_context() -> RequestContext:
    return RequestContext(
        user_id="demo-user",
        tenant_id="demo-tenant",
    )

If the logging is not centralized at the beginning, then it spreads through the “print()” and random “basicConfig”. For a starter kit, one place is enough to configure the format, level, and correlation fields like ‘request id’ and ‘thread id’.

This distinction is very disciplined. “RequestContext” contains immutable launch data: who is the user, which tenant, what rights. AgentState is an evolving state of execution. If you mix them, the count will begin to live with too many duties at once.

context = who started the run and what conditions
What happened inside the run?

The state of the graph should be short, understandable and suitable for serialization. For a starter kit, you almost always need “messages”, a pair of domain fields and a flag for human review.

from typing_extensions import TypedDict, Annotated
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    answer: dict | None
    review_required: bool | None
    metadata: dict

This is where the model, tools and middleware live. Even if you have orchestration by hand on 'LangGraph', this layer is conveniently assembled according to the canons of LangChain: 'init chat model', '@tool', 'ToolRuntime', '@wrap tool call', '@dynamic prompt' and structured output contracts.

@tool
def get_profile(runtime: ToolRuntime[RequestContext]) -> str:
    user_id = runtime.context.user_id
    return f"profile for {user_id}"

Even in the starter kit, it is useful to write nodes “adult” right away: read the state, called the internal component, returned the update. This facilitates replay, persistence and testing.

async def agent_node(state: AgentState, runtime):
    result = await inner_agent.ainvoke(
        {"messages": state["messages"]},
        context=runtime.context,
    )
    return {
        "messages": result["messages"],
        "answer": result.get("structured_response"),
    }

Starter kit routing logic should also be isolated. If you need a review, go to the review; if not, finish. This small function gives a lot of order when the graph starts to grow.

from langgraph.graph import END

def route_after_agent(state: AgentState):
    if state.get("review_required"):
        return "review"
    return END

This is where you collect state schema, nodes, routes and persistence. The cleaner the file, the easier it is to replace models, tools and nodes without breaking the architecture. In starter kit, it’s helpful to discipline yourself with this rule.

def build_graph(settings, checkpointer, store):
    builder = StateGraph(AgentState)
    builder.add_node("agent", agent_node)
    builder.add_node("review", review_node)
    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", route_after_agent)
    builder.add_edge("review", END)
    return builder.compile(checkpointer=checkpointer, store=store)

The service layer is designed to keep routers unaware of the 'RunnableConfig', 'thread id', 'Command(resume=...)' and shape of the final answer. This location links external request/response schemas and internal runtime.

class ChatService:
    def __init__(self, graph, settings):
        self.graph = graph
        self.settings = settings

    async def chat(self, req: ChatRequest, ctx: RequestContext):
        config = {"configurable": {"thread_id": req.thread_id or new_thread_id()}}
        result = await self.graph.ainvoke(
            {"messages": [{"role": "user", "content": req.message}]},
            config=config,
            context=ctx,
        )
        return to_chat_response(result, config)

Streaming is better not to hide inside the usual chat service. A separate service helps neatly form a generator, collect several ‘stream mode’ and decide how to code out tokens, updates and interrupts.

async for item in graph.astream(
    inputs,
    config=config,
    context=ctx,
    stream_mode=["messages", "updates"],
):
    yield encode_sse(item)

The minimum set of public APIs for this service is very specific: "POST /v1/chat", "POST /v1/chat/stream", "POST /v1/threads/{thread id}/resume". That’s enough for a frontend, another service, or CLI to live with.

API SURFACE

POST /v1/chat
POST /v1/chat/stream
POST /v1/threads/{thread_id}/resume
GET  /healthz
GET  /readyz

A common mistake is to type only the usual answer. But if you have a human review, you also have an interrupt contract. It is also better to formalize a separate scheme: pause type, message, action schema, resume payload.

class ChatRequest(BaseModel):
    thread_id: str | None = None
    message: str

class ResumeRequest(BaseModel):
    payload: dict

class InterruptPayload(BaseModel):
    type: str
    message: str
    data: dict

‘healthz’ can simply say ‘the process is alive’. “readyz” is more useful to do a little smarter: check that the service object is collected, persistence is available and basic dependencies have risen. For starter kit, a simple but honest readiness check is enough.

Even if you use in-memory mode in local dev, it is useful to start a single module in the starter kit that can collect persistence resources. Then the path to durable execution in staging/prod remains direct, not chaotic.

async def build_persistence(settings):
    if settings.environment == "local":
        return in_memory_checkpointer(), in_memory_store()
    return durable_checkpointer(settings), durable_store(settings)

FastAPI docs are a distinct reminder: in orchestrators like Kubernetes, one Uvicorn process per container and replica scaling is often preferred over a complex multiprocess inside. This simplifies memory model and lifecycle applications.

You don't need platform art. More important is a fast, clear image that puts dependencies, copies the code and runs Uvicorn. Difficulty can always be added, but the chaos in the container from the beginning only gets in the way.

FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml .
RUN pip install .
COPY app ./app
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

If the old 'thread id' doesn't resolve after the restart, you don't have a service yet, you have a demo. Even a minimal production-oriented build should be built so that the conversation state can continue after the process is upgraded or dropped.

The minimum working set of tests should cover several layers. API tests verify contracts. Service tests check the conversion of HTTP payload to runtime call. Graph tests test routing and persistent behavior. A separate test is required for “interrupt/resume”.

TEST SET

test_chat_route.py
test_resume_route.py
test_chat_service.py
test_graph_flow.py
test_interrupt_resume.py

If later you want to go to the MCP server instead of direct SDKs, it is more convenient to provide a separate module-adapter in advance. Then LangChain tools continue to look the same, and transport and protocol logic does not follow in graph code.

tool
→ repository / adapter
→ mcp_client
→ remote MCP server

You can add an MCP server if you want an IDE, external hosts, or other agents to use your capabilities. But this is the second interface of the product, not a mandatory part of the basic HTTP service.

mcp = FastMCP("starter-service")

@mcp.tool
def search_kb(query: str) -> str:
    ...

@mcp.resource("thread://{thread_id}")
def get_thread(thread_id: str) -> str:
    ...

First, you need to make a minimum durable service. Then you add better auth, observability, queue for long jobs, MCP adapters, richer tools, more complex graph, evals and rate limiting. But the foundation remains the same.

1. basic chat / stream / resume
2. durable persistence
3. auth + readiness + structured logs
4. background jobs + queue
5. MCP client/server if needed
6. evals, guardrails, tracing, budgets

The most common problems are predictable: giant `main.py`, graph compilation for each request, state only in RAM, no resume endpoint, routes with business logic, tools with global singletons and zero tests on the path.

A good starter kit for ‘FastAPI + LangGraph + LangChain’ is not a set of fun libraries, but a small backend with the right boundaries. If you have a ‘lifespan’, ‘app.state’, typed contracts, compiled graph, service layer, durable persistence and an honest ‘chat/stream/resume’ API, then you no longer have a demo, but the basis of a live external service.

Summary formula

FastAPI
  + clean repo layout
  + LangChain components
  + LangGraph runtime
  + external persistence
  + chat/stream/resume contract
  =
Starter kit that can be developed in production service