Streaming LLM Responses with FastAPI: Tokens and Tool Calls

An agent that calls tools can take ten seconds or more to answer, and a blank screen for ten seconds feels broken. Streaming fixes that: the page shows "Getting order status…" the moment the agent calls a tool, then the result, then the answer. In this guide you'll stream an AI agent's response from FastAPI to a web page with Server-Sent Events: tool-call events, tokens, cancellation when the user leaves, and errors halfway through. Every snippet ran against Promptise Foundry 1.3.1 with gpt-5-mini, and every output is pasted from a real run.

Return a StreamingResponse with the media type text/event-stream, and write each event from your agent as a data: line followed by a blank line. That format is Server-Sent Events (SSE), and every browser can read it with EventSource. With Promptise Foundry, agent.astream_with_tools() gives you the events: one when a tool starts, one when it ends, one per token, and one when the agent is done.

@app.get("/chat")
async def chat(q: str):
    async def sse():
        async for event in app.state.agent.astream_with_tools(
            {"messages": [{"role": "user", "content": q}]}
        ):
            yield f"data: {event.to_json()}\n\n"

    return StreamingResponse(
        sse(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Tool events stream live, and so do the tokens of the answer, before and after a tool call.

The browser opens one long HTTP response. Your endpoint loops over the agent's events and writes each one as soon as it exists. Nothing waits for the whole run to finish. The model's answer streams as token events before done, whether or not it called a tool first.

These are the methods on the agent, printed from the installed package:

astream (self, input: 'Any', config: 'dict[str, Any] | None' = None, *, caller: 'CallerContext | None' = None, **kwargs: 'Any') -> 'AsyncIterator[Any]'
astream_with_tools (self, input: 'Any', config: 'dict[str, Any] | None' = None, *, caller: 'CallerContext | None' = None, include_arguments: 'bool' = True, tool_display_names: 'dict[str, str] | None' = None, **kwargs: 'Any') -> 'AsyncIterator[Any]'

Use astream_with_tools() for a page like this. astream() works as well, but it yields the agent's message list after each step, not typed events.

astream_with_tools() yields event objects. All of them share type and timestamp from the StreamEvent base class, and to_json() turns any of them into one SSE-ready line:

timestamp comes from the server's monotonic clock, so it's only useful for measuring the time between two events.

• Python 3.10 or newer. This guide ran on Python 3.12 with FastAPI 0.143.0 and uvicorn 0.54.0.

• Promptise Foundry 1.3.1 or newer, FastAPI and uvicorn: pip install "promptise>=1.3.1" fastapi "uvicorn[standard]"

• An OpenAI API key in OPENAI_API_KEY. Other providers work by changing the model string, as listed in Models & Providers.

• curl, to watch the raw stream.

You'll build a support assistant with one MCP tool that looks up orders, put it behind a FastAPI endpoint, and show its progress on a page.

Real tools take time, and that time is what the user stares at. This tool sleeps for two seconds to stand in for a slow warehouse API, and prints to stderr when it starts and finishes so you can see it working from the server log.

import asyncio
import sys

from promptise.mcp.server import MCPServer, ToolError

server = MCPServer("orders")

# A stand-in for your real database or API.
ORDERS = {
    "A-1001": {"status": "shipped", "carrier": "DHL", "eta": "2026-10-14"},
    "A-1002": {"status": "processing", "carrier": None, "eta": None},
}


@server.tool()
async def get_order_status(order_id: str) -> dict:
    """Get an order's status, carrier and expected delivery date.

    Args:
        order_id: The order ID, for example "A-1001".
    """
    print(f"[orders] lookup {order_id} started", file=sys.stderr, flush=True)
    await asyncio.sleep(2)  # pretend the warehouse API is slow
    order = ORDERS.get(order_id.strip().upper())
    if order is None:
        raise ToolError(f"No order found with ID {order_id}.")
    print(f"[orders] lookup {order_id} finished", file=sys.stderr, flush=True)
    return {"order_id": order_id.strip().upper(), **order}


if __name__ == "__main__":
    server.run()

If MCP servers are new to you, How to Connect MCP Servers to Your AI Agent in Python explains each part of this file.

Build the agent once when the app starts, not once per request: connecting to MCP servers takes time, and one agent can serve many requests at once. Two streams I ran side by side each got their own tool events and their own answer. FastAPI's lifespan is the place for that.

import sys
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

from promptise import build_agent
from promptise.config import StdioServerSpec

HERE = Path(__file__).parent


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Build the agent once at startup and share it between requests.
    app.state.agent = await build_agent(
        model="openai:gpt-5-mini",
        servers={
            "orders": StdioServerSpec(command=sys.executable, args=[str(HERE / "orders_server.py")]),
        },
        instructions="You are a friendly support assistant. Use your tools to answer questions about orders.",
    )
    yield
    await app.state.agent.shutdown()


app = FastAPI(lifespan=lifespan)


@app.get("/chat")
async def chat(q: str):
    async def sse():
        async for event in app.state.agent.astream_with_tools(
            {"messages": [{"role": "user", "content": q}]}
        ):
            yield f"data: {event.to_json()}\n\n"

    return StreamingResponse(
        sse(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Three choices here are deliberate:

• The endpoint is a `GET`. The browser's EventSource can only send GET requests, so the question travels in the query string.

• Each event ends with a blank line. That's how SSE separates events. Forget the second \n and the browser never fires a message.

• Two headers stop caching and buffering. Cache-Control: no-cache keeps caches out of the way, and X-Accel-Buffering: no tells nginx not to hold the response back. More on that under Before you go to production.

Start the app, then call it from a second terminal. -N tells curl to print data as it arrives instead of buffering it:

uvicorn app:app --port 8220
curl -sN "http://127.0.0.1:8220/chat?q=Where%20is%20my%20order%20A-1001%3F"

data: {"type": "tool_start", "timestamp": 49747.397333875, "tool_name": "get_order_status", "tool_display_name": "Getting order status", "arguments": {"order_id": "A-1001"}, "tool_index": 0}

data: {"type": "tool_end", "timestamp": 49749.461395166, "tool_name": "get_order_status", "tool_summary": "order_id: A-1001, status: shipped, carrier: DHL (+1 more)", "duration_ms": 2062.3, "success": true, "tool_index": 0}

data: {"type": "token", "timestamp": 49751.455907916, "text": "Your", "cumulative_text": "Your"}

data: {"type": "token", "timestamp": 49751.530242708, "text": " order", "cumulative_text": "Your order"}

…
data: {"type": "token", "timestamp": 49751.994420291, "text": "?", "cumulative_text": "Your order A-1001 has shipped via DHL and is expected to arrive on 2026-10-14 (in about 3 days). \n\nWould you like me to:\n- fetch the tracking number/link, or\n- request a delivery change, or\n- contact support about this shipment?"}

data: {"type": "done", "timestamp": 49752.051558916, "full_response": "Your order A-1001 has shipped via DHL and is expected to arrive on 2026-10-14 (in about 3 days). \n\nWould you like me to:\n- fetch the tracking number/link, or\n- request a delivery change, or\n- contact support about this shipment?", "tool_calls": [{"name": "get_order_status", "summary": "order_id: A-1001, status: shipped, carrier: DHL (+1 more)", "success": true}], "duration_ms": 6597.4, "cache_hit": false}

The tool events arrive the moment they happen, each with a ready-made label (Getting order status) and a one-line summary of the result. That part is exactly what a progress UI needs.

Then the answer streams as token events, and done carries the whole answer in full_response, with the tool calls that led to it.

Ask something that needs no tool, and tokens do stream:

curl -sN "http://127.0.0.1:8220/chat?q=Hi%21%20In%20one%20sentence%2C%20what%20can%20you%20help%20me%20with%3F"

data: {"type": "token", "timestamp": 49753.641397708, "text": "I", "cumulative_text": "I"}

data: {"type": "token", "timestamp": 49753.646388083, "text": " can", "cumulative_text": "I can"}

data: {"type": "token", "timestamp": 49753.650363041, "text": " help", "cumulative_text": "I can help"}

…
data: {"type": "token", "timestamp": 49753.7964035, "text": ".", "cumulative_text": "I can help with order status and delivery updates, answer product and account questions, troubleshoot issues, and guide you through returns, refunds, and other support needs."}

data: {"type": "done", "timestamp": 49753.815026208, "full_response": "I can help with order status and delivery updates, answer product and account questions, troubleshoot issues, and guide you through returns, refunds, and other support needs.", "tool_calls": [], "duration_ms": 1628.4, "cache_hit": false}

text is the new piece and cumulative_text is everything so far, so a client that misses an event can still redraw the full text. And done follows the last token straight away: a streamed run makes one model call per step, so there's nothing left to wait for.

The stream already carries everything the page needs, so the only change to app.py is a handler that notices when the client goes away, which Step 6 uses:

@app.get("/chat")
async def chat(q: str):
    async def sse():
        try:
            async for event in app.state.agent.astream_with_tools(
                {"messages": [{"role": "user", "content": q}]}
            ):
                yield f"data: {event.to_json()}\n\n"
        except asyncio.CancelledError:
            print(f"client disconnected, agent run stopped: {q!r}", flush=True)
            raise

    return StreamingResponse(
        sse(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Add import asyncio at the top of the file. Run the same question again, this time with -i to see the headers:

curl -isN "http://127.0.0.1:8220/chat?q=Where%20is%20my%20order%20A-1001%3F"

HTTP/1.1 200 OK
date: Sun, 11 Oct 2026 11:14:32 GMT
server: uvicorn
cache-control: no-cache
x-accel-buffering: no
content-type: text/event-stream; charset=utf-8
transfer-encoding: chunked

data: {"type": "tool_start", "timestamp": 49789.786350458, "tool_name": "get_order_status", "tool_display_name": "Getting order status", "arguments": {"order_id": "A-1001"}, "tool_index": 0}

data: {"type": "tool_end", "timestamp": 49792.061326083, "tool_name": "get_order_status", "tool_summary": "order_id: A-1001, status: shipped, carrier: DHL (+1 more)", "duration_ms": 2274.6, "success": true, "tool_index": 0}

…

The headers are the ones the endpoint set, and the tool events come first; the answer's token events and done follow them. The page in the next step handles both: it appends tokens as they come, and replaces the text with full_response at the end, so it shows the whole answer even if it missed a token.

The browser side is plain JavaScript. EventSource opens the stream, and one handler sorts events by type: tool events become a list of steps, tokens are appended to the answer, and done or error ends it.

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Order assistant</title>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 40rem; margin: 2rem auto; padding: 0 1rem; }
    #steps { color: #555; font-size: 0.9rem; padding-left: 1.2rem; }
    #answer { white-space: pre-wrap; line-height: 1.5; }
  </style>
</head>
<body>
  <form id="ask">
    <input id="q" size="40" value="Where is my order A-1001?">
    <button>Ask</button>
  </form>
  <ul id="steps"></ul>
  <div id="answer"></div>

  <script>
    const steps = document.getElementById("steps");
    const answer = document.getElementById("answer");
    let source = null;

    document.getElementById("ask").addEventListener("submit", (e) => {
      e.preventDefault();
      if (source) source.close();
      steps.innerHTML = "";
      answer.textContent = "";

      const q = document.getElementById("q").value;
      source = new EventSource("/chat?q=" + encodeURIComponent(q));

      source.onmessage = (msg) => {
        const event = JSON.parse(msg.data);
        if (event.type === "tool_start") {
          const li = document.createElement("li");
          li.textContent = event.tool_display_name + "…";
          steps.appendChild(li);
        } else if (event.type === "tool_end") {
          const li = document.createElement("li");
          li.textContent = "Done: " + event.tool_summary;
          steps.appendChild(li);
        } else if (event.type === "token") {
          answer.textContent += event.text;
        } else if (event.type === "done") {
          answer.textContent = event.full_response;
          source.close();
        } else if (event.type === "error") {
          answer.textContent = event.message;
          source.close();
        }
      };

      // Network failure: stop instead of letting the browser retry the question.
      source.onerror = () => source.close();
    });
  </script>
</body>
</html>

Serve it from the same app, so the page and the stream share an origin and you need no CORS setup. Add this route to app.py, and HTMLResponse to the fastapi.responses import:

@app.get("/")
async def index():
    return HTMLResponse((HERE / "index.html").read_text())

curl -is http://127.0.0.1:8220/

HTTP/1.1 200 OK
date: Sun, 11 Oct 2026 11:15:07 GMT
server: uvicorn
content-length: 1950
content-type: text/html; charset=utf-8

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
…

Open http://127.0.0.1:8220/ in your browser and ask a question.

The source.close() calls matter more than they look. When a stream ends, EventSource reconnects on its own after a few seconds, and here a reconnect means the same question goes to the agent again. I checked with Node's built-in EventSource (Node 24, behind --experimental-eventsource), which follows the same standard, connected to this endpoint and never closed:

0.1s connection opened (#1)
1.3s tool_start
3.4s tool_end
4.2s done
4.2s stream ended, readyState = 0
7.3s connection opened (#2)
7.3s reconnected and asked the same question again; closing

Three seconds after done, it was back and the agent started a second run. Close the source on done and on error, every time.

When a user closes the tab, there's no point paying for the rest of the run. With uvicorn, Starlette notices the disconnect and cancels your generator, which raises asyncio.CancelledError inside the async for loop. Step 4's except block logs it and re-raises, so the cancellation carries on through the agent.

To test it, disconnect curl after three seconds, while the tool is still running:

curl -sN --max-time 3 "http://127.0.0.1:8220/chat?q=Where%20is%20my%20order%20A-1002%3F"

The server log:

INFO:     127.0.0.1:65242 - "GET /chat?q=Where%20is%20my%20order%20A-1002%3F HTTP/1.1" 200 OK
[orders] lookup A-1002 started
client disconnected, agent run stopped: 'Where is my order A-1002?'
[orders] lookup A-1002 finished

Two things to read from this. The agent run stopped: I also cancelled a run in a script right after tool_start and watched the model callbacks for ten more seconds, and no further model call started. But the tool on the MCP server still ran to the end. Cancelling the agent doesn't stop work that's already running on the server, so tools that change data should finish safely even when nobody is waiting for them. The next request on the same agent worked normally.

Once the first event is out, the HTTP status is already 200 OK, so failures have to travel as events. Promptise has two shapes for them.

A tool that reports an error arrives as a tool_end with success set to false and the server's message as the summary. Here the order doesn't exist and the tool raised ToolError:

data: {"type": "tool_end", "timestamp": 49854.578449333, "tool_name": "get_order_status", "tool_summary": "No order found with ID Z-9999.", "duration_ms": 2069.9, "success": false, "tool_index": 0}

The server marks the result as an error, so your UI can mark the step as failed. The model still gets the message and goes on to tell the user the order wasn't found. A tool that raises an unexpected exception on a Promptise MCP server reports An internal error occurred., and a tool that raises inside the agent's own process gets the summary Tool call failed, so no exception text reaches the browser.

When the run itself fails, for example because the model provider rejects the request, you get an error event and no done. To produce one, I started a second copy of the app with an invalid API key:

data: {"type": "error", "timestamp": 50013.069764958, "message": "An error occurred during processing.", "recoverable": false}

The message is deliberately generic, so no internal detail reaches the browser. The detail goes to your server log instead: Promptise logs a failed streamed run as one ERROR line on the promptise.agent logger, with the error and the agent, session and invocation IDs. To log the provider's error from the model call in your own words too, add a callback, passed in config like any LangChain callback (config={"callbacks": [LogModelErrors()]}):

import logging

from langchain_core.callbacks import AsyncCallbackHandler

log = logging.getLogger("app")


class LogModelErrors(AsyncCallbackHandler):
    """Send the provider's error to your log; the browser only gets the generic message."""

    async def on_llm_error(self, error, **kwargs):
        log.error("model call failed: %s", error)

With the invalid key, and logging set up with logging.basicConfig(format="%(name)s %(levelname)s: %(message)s"), the server logged both:

app ERROR: model call failed: Error code: 401 - {'error': {'message': 'Incorrect API key provided: sk-not-a*****-key. You can find your API key at https://platform.openai.com/account/api-keys.', 'type': 'invalid_request_error', 'code': 'invalid_api_key', 'param': None}, 'status': 401}
promptise.agent ERROR: Streamed run failed: GraphExecutionError: Graph 'react' failed at node 'reason': OpenAIAuthenticationError: Error code: 401 - {'error': {'message': 'Incorrect API key provided: sk-not-a*****-key. You can find your API key at https://platform.openai.com/account/api-keys.', 'type': 'invalid_request_error', 'code': 'invalid_api_key', 'param': None}, 'status': 401} (agent=openai:gpt-5-mini, session=None, invocation=52de3941d407f4fb)

Show the generic message to the user, close the EventSource, and keep the log for yourself.

The automatic labels come from the tool name: get_order_status becomes Getting order status. To say something your users understand, pass tool_display_names. To keep tool arguments out of the browser, pass include_arguments=False:

async for event in agent.astream_with_tools(
    {"messages": [{"role": "user", "content": "Where is my order A-1001?"}]},
    tool_display_names={"get_order_status": "Checking with the warehouse"},
    include_arguments=False,
):
    if event.type == "tool_start":
        print(event.to_json())

{"type": "tool_start", "timestamp": 50158.570676416, "tool_name": "get_order_status", "tool_display_name": "Checking with the warehouse", "arguments": {}, "tool_index": 0}

Hiding arguments is worth considering for any public page. They often carry customer IDs, emails or search terms that the person looking at the screen doesn't need.

Streaming makes waiting visible. It doesn't make the work faster. To see where the time goes, this small client prints when each event arrives:

"""Print when each server-sent event arrives, in seconds since the request started."""

import json
import sys
import time

import httpx

start = time.monotonic()
tokens, last_token = 0, 0.0
with httpx.stream("GET", "http://127.0.0.1:8220/chat", params={"q": sys.argv[1]}, timeout=None) as response:
    for line in response.iter_lines():
        if not line.startswith("data: "):
            continue
        event = json.loads(line.removeprefix("data: "))
        elapsed = time.monotonic() - start
        if event["type"] == "token":
            tokens += 1
            last_token = elapsed
            if tokens == 1:
                print(f"{elapsed:5.1f}s  first token")
            continue
        if tokens:
            print(f"{last_token:5.1f}s  last token ({tokens} tokens)")
            tokens = 0
        detail = event.get("tool_display_name") or event.get("tool_summary") or event.get("message") or ""
        print(f"{elapsed:5.1f}s  {event['type']}  {detail}".rstrip())

With a tool call:

  1.5s  tool_start  Getting order status
  3.6s  tool_end  order_id: A-1001, status: shipped, carrier: DHL (+1 more)
  4.2s  first token
  4.4s  last token (24 tokens)
  4.4s  done

Without one:

  1.3s  first token
  1.5s  last token (23 tokens)
  1.5s  done

Read the first run as three waits. For a second and a half nothing arrives: the model is deciding to call a tool, and tool_start only fires once it has finished writing the call. Then the tool's own two seconds. Then a short pause while the model starts its answer, which streams as tokens, with done right after the last one. Only the first wait is silent. For it, show a "Thinking…" state as soon as the request goes out; no event will come sooner.

In the second run, the first token arrived after 1.3 seconds and done came with the last one, so a UI that waits for done before it re-enables the input box doesn't wait for nothing.

Streaming in Promptise Foundry 1.3.1 has a few things you should know before you build on it:

• The first wait is silent. No event arrives while the model decides to call a tool; tool_start fires once the call is written.

• Text before a tool call streams too. If the model writes something and then calls a tool, those tokens arrive and are in cumulative_text, but full_response on done holds only the final answer. Replace the text with full_response at the end, as the page does.

• Cancelling the agent doesn't stop a tool that's already running. The MCP server finishes the call, as the Step 6 log shows. Make tools that change data safe to finish with nobody waiting.

• The browser gets a generic error. An error event says only "An error occurred during processing." The cause is in your server log, from Promptise's promptise.agent logger or your own callback.

• Turn off proxy buffering. nginx buffers responses from upstream servers by default, which turns your stream into one late chunk. The X-Accel-Buffering: no header switches that off for this response; see proxy_buffering in the nginx docs. Other proxies and CDNs have their own settings. If curl streams fine against uvicorn but not through your proxy, buffering is the first suspect.

• Keep the agent shared, and the request separate. Build one agent at startup, as in Step 2. Pass caller= to astream_with_tools() when your users need their own identity; Multi-user systems covers how.

• Mind the URL. EventSource sends the question in the query string, and query strings end up in access logs: uvicorn's own log above prints every question. If questions can contain personal data, keep the stream on a GET that carries a short-lived ID, and send the question itself in a POST first.

• Plan for idle time. A run can go quiet for several seconds while the model thinks. Check that your proxy and load balancer timeouts are longer than your slowest run.

How do I stream an LLM response to the frontend?

Send Server-Sent Events from your backend and read them with EventSource in the browser. On the server, that's a StreamingResponse with media_type="text/event-stream" that writes data: <json> and a blank line per event. On the page, one onmessage handler appends tokens and shows tool steps, and closes the source when the run is done.

Should I use SSE or WebSockets for streaming LLM responses?

SSE, unless you need the client to send messages during the same connection. A streamed answer only flows one way, and SSE does that over plain HTTP, through ordinary proxies, with no extra library in the browser. WebSockets make sense for things like voice or live collaboration, where both sides talk at once.

Can I stream with a POST request?

Not with EventSource, which only sends GET requests. The same endpoint works as a POST if you read the response with fetch() and its body stream instead, and parse the data: lines yourself. You give up the browser's built-in parsing and get a request body in return.

Why does my stream arrive all at once?

Something between the server and the client is buffering it. Test the endpoint directly with curl -N first. If that streams, look at the proxy in front: set X-Accel-Buffering: no for nginx, and check your CDN or load balancer for a buffering option.

Is Promptise a LangChain streaming alternative?

Promptise Foundry uses LangChain's model integrations underneath, so LangChain callbacks still work, as LogModelErrors shows. What it adds for streaming is a small set of typed events, with tool display names, result summaries and argument hiding built in, instead of raw callback events you have to filter yourself.

• Streaming with Tool Visibility: the reference for astream_with_tools() and its events.

• Building Agents: every option on build_agent.

• Guardrails: checks on input and output, including tool arguments in the stream.

• Observability and Events: traces and notifications to run alongside the stream.

• Build a Production MCP Server in Python: make the tools behind your agent safe to call.

• How to Connect MCP Servers to Your AI Agent in Python: the agent and MCP basics this guide builds on.