Mini-Repository: FastAPI + LangGraph + LangChain as a Service

If you remove the excess magic, the external LLM microservice looks like a normal backend. FastAPI accepts queries and gives answers/streams. LangGraph manages stateful execution: threads, routing, interrupts, checkpoints. LangChain provides models, messages, tools, middleware and structured output. Everything else is the infrastructure around this core.

External layer and internal RUNTIME

browser / frontend / other service
            ↓
        FastAPI
  routes / auth / schemas / stream
            ↓
      Agent Service
            ↓
       LangGraph
 state / checkpoints / interrupts
            ↓
      LangChain
 model / tools / middleware
            ↓
 DB / vector DB / queue / object storage / MCP / external APIs

It's not code size or Docker. Microservice is made by an external contract, a separate deploy lifecycle, its own health/readiness rules, its own persistence logic and the ability to safely serve customers from the outside as an independent service.

Almost all production scenarios fit into three forms: the usual synchronous request/response, streaming response, and long-lived flow with interrupt/resume or background processing. If you think through all three modes in advance, the architecture becomes more sustainable.

1. sync:
   I have been waiting for JSON to come back.

2. stream:
   request → on the way to give tokens / updates

3. durable:
   Request → saved state → pause / worker / resume later

The client doesn’t usually need to know which “retrieve node”, “review node” or “tool router” you have. Externally, it is better to give APIs like "chat", "resume review", "search knowledge", "summarize document". The internal graph remains your implementation, not part of a public contract.

As soon as the service supports multi-turn, streaming history or ‘interrupt/resume’, the outside world starts to refer to specific threads. This means that ‘thread id’, ‘run id’ or their counterparts should be designed consciously, rather than generating randomly ‘inside’.

client sends:
  thread_id? maybe existing

service decides:
  new thread or continue old thread

response returns:
  thread_id
  maybe interrupt payload
  maybe run metadata

The purpose of the structure is to separate the public API, LangGraph runtime, LangChain components and infrastructure. Then the service remains readable and does not turn into one giant 'main.py' with routes, prompts, tools and SQL in a jumble.

REPO TREE

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

Graph, checkpointer, store, database pool, model clients and service objects are better collected at the start of the application, not for every request. For this, FastAPI has a ‘lifespan’, and it is one of the most important practical pieces for a production service.

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

app = FastAPI(lifespan=lifespan)

An HTTP route does not need to know how a graph works. Its job is simpler: fail the input, pull out the auth/tenant context, invoke the service method, and return the JSON or stream. This keeps the HTTP code small and replaceable.

Inside the “agent” live domain contracts, models, tools, middleware, graph nodes and service orchestration. This folder is responsible for the business behavior of the application. This is where most of the real value of the service is concentrated.

contracts   -> state / context / response schema
components  -> model / tools / middleware / prompt layer
graph       -> nodes / routes / build
services    -> invoke / stream / resume orchestration

Persistence, queue, MCP server/client glue, metrics, logging setup, and other operational code are growing rapidly. If you do not take them into a separate layer, they will mix with the graph code and begin to complicate both the development and the depot.

If the user needs a regular response without streaming, it is convenient to have one sync endpoint. Inside, it builds a 'thread id', runtime context, calls 'graph.invoke(...)' through the service, and returns a validated 'response model' along with metadata.

@router.post("/v1/chat", response_model=ChatResponse)
async def chat(req: ChatRequest, svc = Depends(get_chat_service), ctx = Depends(get_request_context)):
    return await svc.chat(req, ctx)

LangGraph can stream tokens and state updates. Outwardly, it is convenient to give it through “StreamingResponse”, usually in the format of “text/event-stream” or chunked JSON lines. A separate endpoint makes the contract clearer and doesn’t force sync clients to understand the streaming protocol.

@router.post("/v1/chat/stream")
async def chat_stream(req: ChatRequest, svc = Depends(get_stream_service), ctx = Depends(get_request_context)):
    return StreamingResponse(
        svc.stream(req, ctx),
        media_type="text/event-stream",
    )

This is one of the most common flaws. People add approval-node with ‘interrupt()’, but forget to design the external path of continuation. Production requires a separate endpoint that accepts 'thread id' and payload for 'Command'.

@router.post("/v1/threads/{thread_id}/resume")
async def resume(thread_id: str, req: ResumeRequest, svc = Depends(get_chat_service), ctx = Depends(get_request_context)):
    return await svc.resume(thread_id=thread_id, payload=req.payload, ctx=ctx)

A live service needs service entry points. Liveness says “process is alive,” readiness says “you can take traffic,” version says “what’s locked in right now,” OpenAPI says “how customers integrate.” These endpoints are quickly becoming a must for the platform team.

If a service lives on the outside, it almost always needs API keys, bearer tokens or other auth-mechanism. It usually gives rise to the request context: “user id”, “tenant id”, roles, feature flags. For browser clients, CORS is added. All this is more convenient to decorate the dependencies layer from FastAPI.

Authorization header
→ verify principal
→ build RequestContext(user_id, tenant_id, scopes)
Put it in the service/graph context

This is one of the “non-obvious but highly relevant” production units. If the client repeats the request because of the timeout, and inside the graph sends an email, creates a ticket or writes in CRM, without idempotence, you will easily get duplicate actions.

The internal runtime should be typed as well as the external HTTP API. “state” describes the living state of the graph, “context” – immutable launch data like “user id” and tenancy, “schemas” – the boundaries of structured output and API. This makes graph code less fragile and easier to test.

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    retrieved_docs: list[str]
    final_answer: dict | None
    approval_required: bool | None

Even if you have orchestration on LangGraph, LangChain is still very appropriate as a layer of finished components. It is home to ‘init chat model’, tools with ‘ToolRuntime’, middleware for model/tool hooks, messages and structured output schemes.

model       = init_chat_model(...)
tools       = [search_docs, create_ticket, get_profile]
middleware  = [dynamic_prompt, tool_error_handler]
response_schema = AnswerSchema

`builder.compile(...)' assembles runnable graph with checkpointer and store. If you do this for every call, you’ll get extra complexity, lost productivity, and unstable resource initialization. The service usually receives a compiled graph from lifespan.

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

This is the place where HTTP payloads are born ‘inputs’, ‘config’ and runtime ‘context’, and then called ‘graph.invoke(...)’ or ‘graph.astream(...)’. Such a service layer is useful because it is convenient to hold thread policy, metadata, error mapping and response shaping.

config = {"configurable": {"thread_id": thread_id}}
context = RequestContext(...)

result = await graph.ainvoke(
    {"messages": [{"role": "user", "content": req.message}]},
    config=config,
    context=context,
)

Once the graph can settle for human approval, the service should be able to beautifully return the payload to the UI or external client. This means a well-thought-out JSON contract: interrupt type, question text, draft answer, acceptable actions, and the way to resume.

One of the most important production shifts: InMemorySaver and local dicts are well suited only for laptop and local development. A live service that restarts, scales and maintains multiple instances must store thread state and memory in external persistence.

Proper separation

app instance A  ─┐
app instance B  ─┼──→ external checkpointer / store / DB
app instance C  ─┘

Then any copy can continue threading thread id

If one query hits worker A and the next thread id hits worker B, the in-memory state disappears. Therefore, for multi-worker and multi-replica scripts, checkpointer/store should be shared, and service instances should be as stateless as possible.

This is another non-obvious, but really important block. If processing can take minutes, call a lot of tools, do reindex or generate a report, it is better to highlight the background job/worker pattern. HTTP endpoint then creates a run and puts the task in a queue, and the client then reads the status or streams.

POST /v1/jobs
→ validate request
→ create run record
→ enqueue task
→ return 202 Accepted + run_id

worker
→ loads same graph/service layer
→ executes durable flow
→ writes status/results

Chats, traces, prompt snapshots, uploaded files, generated reports and other useful artifacts should be stored in an external storage: SQL / NoSQL, object storage, vector DB. The container or process does not have to be the only place where user data lives.

The real service has limits: upstream LLM timeout, tool timeout, client timeout, ingress timeout, limit on simultaneous streams. If they are not designed in advance, the service will begin to hang, accumulate hanging connections and fall under load.

One of the most useful shifts in understanding is that FastAPI and MCP are different. A public HTTP API is needed for browsers, backends, and business integrations. MCP is needed where LLM hosts and agent systems want standardized tools, resources, and prompts. It's not the same thing.

Two adjacent layers

public app API:
  browser / backend / mobile → FastAPI

tool/data plane:
  host / agent platform / IDE → MCP server

Often the same product has both layers.

This is a good pattern if your LangGraph service needs standardized tools from the outside world: GitHub, Slack, files, CRM, internal knowledge, etc. Then your ‘LangChain’ tool layer can run through MCP clients rather than holding dozens of custom SDKs.

FastAPI request
→ graph node
→ LangChain tool
→ MCP client
→ remote MCP server
→ tool result back into graph

If your product itself knows useful operations or has important data, you can add “mcp server.py” and give some of the features out through MCP. Then IDEs, external hosts or other agent platforms will be able to use your tools/resources/prompts in a standard way.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-agent-platform")

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

@mcp.resource("kb://policy/{name}")
def get_policy(name: str) -> str:
    ...

It is better not to confuse these interfaces. FastAPI is good as a business API for applications and services. MCP is a standardized tool/data protocol for hosts and agents. Very often, the correct answer is to have both interfaces, but with different audiences and different areas of responsibility.

If you are picking up a remote MCP server, think not only about transport, but also about auth, origin checks, tool descriptions, prompt injection through resources and rights to side-effect tools. An MCP server is a login to your tool/data plane, not just another JSON endpoint.

Logs “received the request” and “response” are not enough. For an LLM service, it is very useful to carry correlation identifiers through the entire stack: HTTP request id, thread id, run id, tool call id. Then you can associate a specific client failure with a specific graph run and a specific execution tool.

request_id  -> HTTP request
thread_id   -> conversation / state line
run id -> specific graph launch
tool call id -> external tool invocation

Even if the endpoint is called the same, the behavior of the service may change due to a new prompt, new tool rules, different routing logic, or other output schema. Therefore, it is useful to have a clear version of the API, prompt/graph revision and migration discipline for customers.

If you test only the endpoint, it will not be clear if the auth, service wiring, routing graph, or specific tool is broken. Therefore, unit tests tools/middleware, integration tests compiled graph and API tests for schemas, auth and stream/resume endpoints are useful here.

TEST LAYERS

HTTP tests
  ↓
service tests
  ↓
graph integration tests
  ↓
tool / interrupt / resume tests

For external life, the service is usually containerized, raised behind proxy/ingress with HTTPS, throwing env vars and scale policy. An important engineering idea: once the state is outdoors, the application becomes much easier to scale and survive restarts.

local dev
→ Docker image
→ staging with real DB/checkpointer
→ prod behind ingress / load balancer
→ replicas + shared persistence

In many clusters, it is easier to scale replicas than to build complex multiprocesses within a single container.

If you compress everything to the most useful engineering framework, the result is: FastAPI gives an external contract, LangGraph gives a long-lasting runtime, LangChain gives convenient building blocks, external persistence makes the service live, and MCP adds a standardized tool/data interface for other hosts and agents if necessary.

One main picture

public clients
   ↓
FastAPI API
   ↓
service layer
   ↓
LangGraph runtime
   ↓
LangChain components
   ↓
external state + tools + MCP + data plane

Such a service can not only be run locally, but also actually operate.