Vol. 24 · Application Engineering
Mini-Repository: FastAPI + LangGraph + LangChain as a Service
This volume does not answer the question “how to write another agent”, but the question “how to make a live external application”. At the center is one solid Python mini-repo: ‘FastAPI’ as a public HTTP/SSE layer, ‘LangGraph’ as runtime, ‘LangChain’ as a layer of models, tools and middleware, external persistence for threads, scaling, auth, health/readiness and a separate place for ‘MCP’ as a tool/data plane.
The Big Picture of 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.
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.
Source-complete notesminimum set
minimum set
- stable API
- separate deploy
- state/storage
- auth / monitoring / retries
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.
Source-complete notesthree modes
text1. 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 laterThe client usually does not need to know that your graph contains `retrieve_node`, `review_node`, or `tool_router`. Expose business operations such as `chat`, `resume_review`, `search_knowledge`, and `summarize_document`; keep the internal graph an implementation detail rather than part of the public contract.
Source-complete notessignal
signal
- Good boundary: API = business operations
- Leaky boundary: API = internal node names
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’.
Source-complete notespractice
textclient sends:
thread_id? maybe existing
service decides:
new thread or continue old thread
response returns:
thread_id
maybe interrupt payload
maybe run metadataOne Whole Mini-Repo
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.
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.
Source-complete notesskeleton
python@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.
Source-complete notesgood way
good way
- schema validation
- auth dependency
- service call
- complex orchestration logic
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.
Source-complete noteslogic
textcontracts -> state / context / response schema
components -> model / tools / middleware / prompt layer
graph -> nodes / routes / build
services -> invoke / stream / resume orchestrationPersistence, 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.
Source-complete notesoften lives infra
often lives infra
- checkpointer/store builders
- queue adapters
- MCP server bootstrap
- logging/tracing setup
Public HTTP API
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.
Source-complete notesexample
python@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.
Source-complete notesexample
python@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'.
Source-complete notesexample
python@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.
Source-complete notesminimum
minimum
- GET /healthz
- GET /readyz
- GET /version
- GET /openapi.json
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.
Source-complete notesidea
textAuthorization header
→ verify principal
→ build RequestContext(user_id, tenant_id, scopes)
Put it in the service/graph contextThis 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.
Source-complete notesparticularly
particularly
- resume after timeout
- tool with external side effect
- repeat client retries
LangGraph and LangChain Inside the Service
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.
Source-complete notesstate
pythonclass AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
retrieved_docs: list[str]
final_answer: dict | None
approval_required: bool | NoneEven 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.
Source-complete notescomponent layer
textmodel = 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.
Source-complete notesbuild_graph
pythondef 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.
Source-complete noteslogic
typescriptconfig = {"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.
Source-complete notescome back
come back
- thread_id
- interrupt type
- payload for UI
- resume action schema
State, Scaling and Long Life
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.
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.
Source-complete notesred flag
red flag
- thread history only in the RAM process
- Local disk as the only storage
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.
Source-complete notespattern
textPOST /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/resultsChats, 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.
Source-complete notesdataplane
dataplane
- SQL / Postgres
- object storage
- vector DB
- queue broker
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.
Source-complete notesthink through
think through
- request timeout
- tool timeout
- queue depth limit
- cancel on client disconnect
Where is MCP Really Needed?
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.
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.
Source-complete noteswhat it looks like
textFastAPI request
→ graph node
→ LangChain tool
→ MCP client
→ remote MCP server
→ tool result back into graphIf 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.
Source-complete notessketch
sqlfrom 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.
Source-complete notestable
| Interface. | Better for you. |
|---|---|
| FastAPI | frontend, backend, product API, auth, business flows |
| MCP | tools/resources/prompts for hosts and agent systems |
| Both | A platform that people and other agents need. |
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.
Source-complete notesreally important
really important
- Streamable HTTP transport
- auth / OAuth
- least privilege tools
- prompt injection via tool/resource content
Relevant production blocks that are often forgotten
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.
Source-complete notesids
textrequest_id -> HTTP request
thread_id -> conversation / state line
run id -> specific graph launch
tool call id -> external tool invocationEven 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.
Source-complete notesversion
version
- HTTP API version
- response schema
- graph/prompt revision
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.
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.
Source-complete noteswalkway
textlocal 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.
No dead end
Keep moving through the map.
Continue in sequence, switch to a related guide, or return to the seven-track learning map.