Contact

LangChain FastAPI Streaming Tutorial

Building agent backends as HTTP services is a practical pattern in 2026. It lets you keep the agent logic in one place, then reuse it from a web UI, a CLI, a notebook, or an internal automation job. LangChain helps with model, tool, and agent orchestration, while FastAPI gives you a clean async API layer with typed request bodies and predictable responses.

In this tutorial, you will build a small Deep Agent, expose it through a POST endpoint, and support two response modes:

  • JSON chat-completion style output for normal requests
  • SSE streaming output when the client enables streaming

That combination is useful because it gives you one endpoint that works for both classic request/response clients and real-time chat UIs.

What You Will Build

By the end, you will have an API with this behavior:

  • Client sends POST /agent
  • If stream is false, the API returns a chat-completion-like JSON payload
  • If stream is true, the API returns an SSE stream
  • The streaming path uses agent.astream(...)

Requirements

  • Basic Python knowledge
  • Some familiarity with APIs and JSON
  • Python 3.11+
  • An API key for a supported chat model provider
  • No GPU required if you use a hosted model

Step 1: Install the Dependencies

Create a fresh project and install the packages:

mkdir lc-fastapi-agent
cd lc-fastapi-agent

python -m venv .venv
source .venv/bin/activate
pip install "fastapi[standard]" uvicorn langchain langchain-openai deepagents

If you want to use a different provider, swap langchain-openai for the matching integration package.

Also set your model provider key in the environment before starting the app.

Step 2: Create a Small Deep Agent

For the tutorial, we will build a small bioinformatics-flavored assistant with a few tools:

  • reverse complement for DNA
  • GC content calculation
  • a tiny gene summary lookup

Create app/main.py and start with this:

import json
import time
from typing import Any, AsyncIterator
from uuid import uuid4

from deepagents import create_deep_agent
from fastapi import FastAPI
from fastapi.responses import JSONResponse, StreamingResponse
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from pydantic import BaseModel, Field

MODEL_NAME = "openai:gpt-5.5"


@tool
def reverse_complement(sequence: str) -> str:
    """Return the reverse complement of a DNA sequence."""
    cleaned = "".join(base.upper() for base in sequence if base.upper() in {"A", "C", "G", "T"})
    if not cleaned:
        return "No valid DNA bases found. Use only A, C, G, or T."

    table = str.maketrans("ACGT", "TGCA")
    return cleaned.translate(table)[::-1]


@tool
def gc_content(sequence: str) -> str:
    """Calculate GC content for a DNA sequence."""
    cleaned = "".join(base.upper() for base in sequence if base.upper() in {"A", "C", "G", "T"})
    if not cleaned:
        return "No valid DNA bases found. Use only A, C, G, or T."

    gc = sum(base in {"G", "C"} for base in cleaned)
    percent = (gc / len(cleaned)) * 100
    return f"Sequence: {cleaned}\nLength: {len(cleaned)}\nGC content: {percent:.2f}%"


@tool
def gene_summary(symbol: str) -> str:
    """Return a small demo gene summary."""
    demo_db = {
        "BRCA1": "BRCA1 is involved in DNA damage repair and is linked to hereditary breast and ovarian cancer risk.",
        "TP53": "TP53 encodes p53, a tumor suppressor that helps coordinate stress responses and genome protection.",
        "EGFR": "EGFR is a receptor tyrosine kinase involved in cell signaling and is a frequent oncology target.",
    }
    return demo_db.get(symbol.upper(), f"No demo summary found for {symbol}.")


model = init_chat_model(MODEL_NAME, temperature=0)

agent = create_deep_agent(
    model=model,
    tools=[reverse_complement, gc_content, gene_summary],
    system_prompt=(
        "You are a careful bioinformatics assistant. "
        "Use tools when they help. "
        "Be concise, correct, and explicit about assumptions."
    ),
)

app = FastAPI(title="LangChain Deep Agent API")

What is happening here?

  • @tool turns normal Python functions into LangChain tools
  • init_chat_model(...) creates the chat model
  • create_deep_agent(...) builds the agent you will expose through FastAPI
  • app = FastAPI(...) creates the HTTP application

This is the core design pattern: agent inside the app, API outside the agent.

Step 3: Define the Request and Response Helpers

Next, add a small request model and helper utilities below the code above:

class AgentRequest(BaseModel):
    message: str = Field(..., min_length=1, description="User prompt")
    stream: bool = Field(default=False, description="Return SSE stream when true")


def extract_final_text(result: dict[str, Any]) -> str:
    messages = result.get("messages", [])

    for message in reversed(messages):
        if getattr(message, "type", None) == "ai":
            text = str(getattr(message, "text", "")).strip()
            if text:
                return text

    return ""


def to_chat_completion(content: str) -> dict[str, Any]:
    return {
        "id": f"chatcmpl-{uuid4().hex}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": MODEL_NAME,
        "choices": [
            {
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": content,
                },
                "finish_reason": "stop",
            }
        ],
    }


def sse_event(data: dict[str, Any], event: str | None = None) -> str:
    lines = []
    if event:
        lines.append(f"event: {event}")
    lines.append(f"data: {json.dumps(data, ensure_ascii=False)}")
    return "\n".join(lines) + "\n\n"

Why this step matters

These helpers keep the endpoint code small and readable:

  • AgentRequest validates the incoming payload
  • extract_final_text(...) pulls the final assistant text from the agent result
  • to_chat_completion(...) gives clients a familiar JSON shape
  • sse_event(...) formats valid SSE chunks

Step 4: Connect the Agent to a POST Endpoint

Now add the non-streaming endpoint behavior.

@app.post("/agent", response_model=None)
async def run_agent(body: AgentRequest):
    if body.stream:
        return StreamingResponse(
            stream_agent(body.message),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": body.message,
                }
            ]
        }
    )

    content = extract_final_text(result)
    return JSONResponse(content=to_chat_completion(content))

What is happening here?

This route does two things:

  • normal mode: agent.ainvoke(...)
  • streaming mode: returns a StreamingResponse

That means you only need one REST endpoint for both interaction styles.

Step 5: Add SSE Streaming with agent.astream

Now define the streaming function used above.

async def stream_agent(message: str) -> AsyncIterator[str]:
    completion_id = f"chatcmpl-{uuid4().hex}"
    created = int(time.time())

    try:
        async for chunk, _metadata in agent.astream(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": message,
                    }
                ]
            },
            stream_mode="messages",
        ):
            text = str(getattr(chunk, "text", ""))
            if not text:
                continue

            yield sse_event(
                {
                    "id": completion_id,
                    "object": "chat.completion.chunk",
                    "created": created,
                    "model": MODEL_NAME,
                    "choices": [
                        {
                            "index": 0,
                            "delta": {"content": text},
                            "finish_reason": None,
                        }
                    ],
                },
                event="delta",
            )

        yield sse_event(
            {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": MODEL_NAME,
                "choices": [
                    {
                        "index": 0,
                        "delta": {},
                        "finish_reason": "stop",
                    }
                ],
            },
            event="delta",
        )

        yield "data: [DONE]\n\n"

    except Exception as exc:
        yield sse_event({"error": str(exc)}, event="error")

Why use agent.astream(...)?

Because it gives you incremental output instead of waiting for the full result. That is exactly what you want for:

  • chat UIs
  • long answers
  • multi-step agent behavior
  • better perceived latency

Why SSE?

SSE is a very clean choice when the server mostly pushes text in one direction. It is simpler than a WebSocket for many agent-chat use cases, especially when the client only needs streamed model output.

Step 6: Review the Complete File

Your full app/main.py should now look like this:

import json
import time
from typing import Any, AsyncIterator
from uuid import uuid4

from deepagents import create_deep_agent
from fastapi import FastAPI
from fastapi.responses import JSONResponse, StreamingResponse
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from pydantic import BaseModel, Field

MODEL_NAME = "openai:gpt-5.5"


@tool
def reverse_complement(sequence: str) -> str:
    """Return the reverse complement of a DNA sequence."""
    cleaned = "".join(base.upper() for base in sequence if base.upper() in {"A", "C", "G", "T"})
    if not cleaned:
        return "No valid DNA bases found. Use only A, C, G, or T."

    table = str.maketrans("ACGT", "TGCA")
    return cleaned.translate(table)[::-1]


@tool
def gc_content(sequence: str) -> str:
    """Calculate GC content for a DNA sequence."""
    cleaned = "".join(base.upper() for base in sequence if base.upper() in {"A", "C", "G", "T"})
    if not cleaned:
        return "No valid DNA bases found. Use only A, C, G, or T."

    gc = sum(base in {"G", "C"} for base in cleaned)
    percent = (gc / len(cleaned)) * 100
    return f"Sequence: {cleaned}\nLength: {len(cleaned)}\nGC content: {percent:.2f}%"


@tool
def gene_summary(symbol: str) -> str:
    """Return a small demo gene summary."""
    demo_db = {
        "BRCA1": "BRCA1 is involved in DNA damage repair and is linked to hereditary breast and ovarian cancer risk.",
        "TP53": "TP53 encodes p53, a tumor suppressor that helps coordinate stress responses and genome protection.",
        "EGFR": "EGFR is a receptor tyrosine kinase involved in cell signaling and is a frequent oncology target.",
    }
    return demo_db.get(symbol.upper(), f"No demo summary found for {symbol}.")


class AgentRequest(BaseModel):
    message: str = Field(..., min_length=1, description="User prompt")
    stream: bool = Field(default=False, description="Return SSE stream when true")


def extract_final_text(result: dict[str, Any]) -> str:
    messages = result.get("messages", [])

    for message in reversed(messages):
        if getattr(message, "type", None) == "ai":
            text = str(getattr(message, "text", "")).strip()
            if text:
                return text

    return ""


def to_chat_completion(content: str) -> dict[str, Any]:
    return {
        "id": f"chatcmpl-{uuid4().hex}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": MODEL_NAME,
        "choices": [
            {
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": content,
                },
                "finish_reason": "stop",
            }
        ],
    }


def sse_event(data: dict[str, Any], event: str | None = None) -> str:
    lines = []
    if event:
        lines.append(f"event: {event}")
    lines.append(f"data: {json.dumps(data, ensure_ascii=False)}")
    return "\n".join(lines) + "\n\n"


model = init_chat_model(MODEL_NAME, temperature=0)

agent = create_deep_agent(
    model=model,
    tools=[reverse_complement, gc_content, gene_summary],
    system_prompt=(
        "You are a careful bioinformatics assistant. "
        "Use tools when they help. "
        "Be concise, correct, and explicit about assumptions."
    ),
)

app = FastAPI(title="LangChain Deep Agent API")


async def stream_agent(message: str) -> AsyncIterator[str]:
    completion_id = f"chatcmpl-{uuid4().hex}"
    created = int(time.time())

    try:
        async for chunk, _metadata in agent.astream(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": message,
                    }
                ]
            },
            stream_mode="messages",
        ):
            text = str(getattr(chunk, "text", ""))
            if not text:
                continue

            yield sse_event(
                {
                    "id": completion_id,
                    "object": "chat.completion.chunk",
                    "created": created,
                    "model": MODEL_NAME,
                    "choices": [
                        {
                            "index": 0,
                            "delta": {"content": text},
                            "finish_reason": None,
                        }
                    ],
                },
                event="delta",
            )

        yield sse_event(
            {
                "id": completion_id,
                "object": "chat.completion.chunk",
                "created": created,
                "model": MODEL_NAME,
                "choices": [
                    {
                        "index": 0,
                        "delta": {},
                        "finish_reason": "stop",
                    }
                ],
            },
            event="delta",
        )

        yield "data: [DONE]\n\n"

    except Exception as exc:
        yield sse_event({"error": str(exc)}, event="error")


@app.post("/agent", response_model=None)
async def run_agent(body: AgentRequest):
    if body.stream:
        return StreamingResponse(
            stream_agent(body.message),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "Connection": "keep-alive",
                "X-Accel-Buffering": "no",
            },
        )

    result = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": body.message,
                }
            ]
        }
    )

    content = extract_final_text(result)
    return JSONResponse(content=to_chat_completion(content))

Step 7: Start the API

Run the server:

uvicorn app.main:app --reload

Step 8: Example Request Shapes

Non-streaming request body

{
  "message": "Explain BRCA1 in two sentences and compute GC content for ACGCGGTTAA.",
  "stream": false
}

Non-streaming response shape

{
  "id": "chatcmpl-123",
  "object": "chat.completion",
  "created": 1760000000,
  "model": "openai:gpt-5.5",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "BRCA1 helps repair DNA damage. GC content for ACGCGGTTAA is 60.00%."
      },
      "finish_reason": "stop"
    }
  ]
}

Streaming request body

{
  "message": "Give the reverse complement of ACGTTGCA and explain each step briefly.",
  "stream": true
}

Streaming SSE shape

event: delta
data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1760000000,"model":"openai:gpt-5.5","choices":[{"index":0,"delta":{"content":"The"},"finish_reason":null}]}

event: delta
data: {"id":"chatcmpl-123","object":"chat.completion.chunk","created":1760000000,"model":"openai:gpt-5.5","choices":[{"index":0,"delta":{"content":" reverse"},"finish_reason":null}]}

data: [DONE]

Why REST Endpoints for Agents Work Well

This pattern is useful because it separates concerns cleanly:

  • FastAPI handles transport: validation, HTTP semantics, auth hooks, deployment
  • LangChain handles reasoning: tools, model calls, orchestration
  • Clients stay simple: they only send JSON and optionally read an SSE stream

That makes the backend easier to test, version, and swap behind multiple frontends.

Recap

You now have a working pattern for serving a LangChain Deep Agent through FastAPI:

  1. Create tools
  2. Build a Deep Agent
  3. Expose one POST route
  4. Use ainvoke for normal JSON responses
  5. Use astream plus StreamingResponse for SSE

This is a strong default architecture for agent APIs because it is simple, production-friendly, and easy to extend with auth, logging, tracing, and persistence.

Further Reading

  • LangChain init_chat_model reference.
  • Deep Agents create_deep_agent reference.
  • LangGraph stream modes reference.
  • FastAPI streaming responses.
  • FastAPI response models.

FAQ

1. Do I need a GPU for this tutorial?

No, not if you use a hosted model provider. A GPU is only relevant if you run the model yourself.

2. Why return chat-completion style JSON instead of plain text?

Because many clients already understand that shape. It also makes it easier to swap this endpoint into existing chat UIs with minimal frontend changes.

3. Why use SSE instead of WebSockets?

If your client mainly needs one-way streamed output from the server, SSE is often simpler. Use WebSockets when you need richer two-way, long-lived interactions.

Related Posts

Leave a Comment