Streaming What Your Research Agent Is Doing, Not Just What It Says

Twenty seconds of blank screen while an agent searches feels broken. Streaming the tool calls turns dead time into visible progress.

Profile picture of Serply
Serply
A UI streaming search progress events before the final answer

Token streaming solved the wrong half of the problem for research agents. The model streams beautifully — but only after it has finished searching, reading four pages, and deciding what to say. The user watches a spinner for twenty seconds, then text appears.

What they should see is: searching for X → found 10 results → reading three sources → writing. Same latency, completely different experience, and it’s the tool calls you need to stream, not just the tokens.

Event shape

Define the events before writing any transport code. They’re your contract with the frontend:

from dataclasses import dataclass, asdict
from typing import Literal
import json


@dataclass
class Event:
    type: Literal["status", "search", "results", "reading", "read_done",
                  "token", "citation", "done", "error"]
    data: dict

    def sse(self) -> str:
        return f"event: {self.type}\ndata: {json.dumps(self.data)}\n\n"

Distinct event types rather than one blob with a kind field means the frontend can bind handlers directly and ignore what it doesn’t render yet.

Emitting around the tool calls

import os
import asyncio
import httpx
from urllib.parse import quote_plus

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"


async def search(client: httpx.AsyncClient, query: str, num: int = 10) -> list[dict]:
    resp = await client.get(
        f"{BASE}/search/q={quote_plus(query)}&num={num}",
        headers={"X-Api-Key": API_KEY, "X-Proxy-Location": "US"},
        timeout=30,
    )
    resp.raise_for_status()
    return [
        {"title": r.get("title", ""), "url": r.get("link", ""),
         "snippet": r.get("description", "")}
        for r in resp.json().get("results", [])
    ]


async def read(client: httpx.AsyncClient, url: str) -> dict:
    resp = await client.post(
        f"{BASE}/request",
        headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
        json={"url": url, "response_type": "markdown"},
        timeout=90,
    )
    resp.raise_for_status()
    return {"url": url, "text": resp.text[:20_000]}

resp.text for markdown mode — the content comes back as the body, not wrapped in JSON.

The pipeline as an async generator

from anthropic import AsyncAnthropic

anthropic = AsyncAnthropic()


async def research_stream(question: str):
    async with httpx.AsyncClient() as client:
        yield Event("status", {"phase": "planning"})

        queries = await plan_queries(question)
        yield Event("status", {"phase": "searching", "queries": queries})

        tasks = [asyncio.create_task(search(client, q)) for q in queries]
        all_results: list[dict] = []

        for query, task in zip(queries, tasks):
            yield Event("search", {"query": query})
            try:
                found = await task
            except Exception as e:
                yield Event("error", {"query": query, "message": str(e)})
                continue
            all_results.extend(found)
            yield Event("results", {
                "query": query,
                "count": len(found),
                "titles": [r["title"] for r in found[:5]],
            })

        seen, urls = set(), []
        for r in all_results:
            if r["url"] and r["url"] not in seen:
                seen.add(r["url"])
                urls.append(r["url"])
        urls = urls[:3]

        pages = []
        read_tasks = {asyncio.create_task(read(client, u)): u for u in urls}
        for u in urls:
            yield Event("reading", {"url": u})

        for task in asyncio.as_completed(read_tasks):
            try:
                page = await task
            except Exception:
                continue
            pages.append(page)
            yield Event("read_done", {"url": page["url"], "chars": len(page["text"])})

        yield Event("status", {"phase": "writing", "sources": len(pages)})

        context = "\n\n".join(f"=== {p['url']} ===\n{p['text']}" for p in pages)
        async with anthropic.messages.stream(
            model="claude-sonnet-4-5",
            max_tokens=2000,
            system=("Answer using only the provided sources. Cite the URL for "
                    "every factual claim. If the sources don't answer the "
                    "question, say so."),
            messages=[{"role": "user",
                       "content": f"Question: {question}\n\nSources:\n{context}"}],
        ) as stream:
            async for text in stream.text_stream:
                yield Event("token", {"text": text})

        yield Event("done", {"sources": [p["url"] for p in pages]})

Three details that matter.

The search event fires before awaiting, so the user sees the query the moment it’s issued rather than when it returns. Firing it after the await would show all three queries appearing simultaneously at the end.

asyncio.as_completed on the reads means each page reports as it lands. Reading three pages sequentially and reporting at the end wastes the whole point — the fastest one should show up in two seconds, not wait for the slowest.

Search failures emit an error event and continue. One dead query shouldn’t kill the run, and the user should see that it happened rather than silently getting a thinner answer.

Serving it

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


@app.get("/research")
async def research(q: str):
    async def gen():
        try:
            async for event in research_stream(q):
                yield event.sse()
        except Exception as e:
            yield Event("error", {"message": str(e), "fatal": True}).sse()

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

X-Accel-Buffering: no is the header people forget. Behind nginx, the default buffers your entire stream and delivers it as one response — the code works perfectly in local development and the streaming silently disappears in production.

The client

const source = new EventSource(`/research?q=${encodeURIComponent(question)}`);
const log = document.getElementById('progress');
const answer = document.getElementById('answer');

source.addEventListener('search', (e) => {
  const { query } = JSON.parse(e.data);
  log.insertAdjacentHTML('beforeend', `<li>Searching: <code>${query}</code></li>`);
});

source.addEventListener('results', (e) => {
  const { query, count } = JSON.parse(e.data);
  log.insertAdjacentHTML('beforeend', `<li>${count} results for “${query}”</li>`);
});

source.addEventListener('reading', (e) => {
  const { url } = JSON.parse(e.data);
  log.insertAdjacentHTML('beforeend',
    `<li data-url="${url}">Reading ${new URL(url).hostname}…</li>`);
});

source.addEventListener('read_done', (e) => {
  const { url } = JSON.parse(e.data);
  const li = log.querySelector(`[data-url="${CSS.escape(url)}"]`);
  if (li) li.textContent = `Read ${new URL(url).hostname}`;
});

source.addEventListener('token', (e) => {
  answer.textContent += JSON.parse(e.data).text;
});

source.addEventListener('done', () => source.close());
source.addEventListener('error', (e) => {
  // EventSource fires a bare error on disconnect too — check for data
  if (e.data) log.insertAdjacentHTML('beforeend',
    `<li class="err">${JSON.parse(e.data).message}</li>`);
});

Showing the hostname rather than the full URL is a small thing that makes the progress log readable. “Reading arxiv.org” scans; “Reading https://arxiv.org/abs/2401.12345v2?utm_source=…” doesn’t.

Don’t call it done at the token stream

One habit worth adopting: keep the progress log visible after the answer arrives, collapsed but expandable. It’s the same information you’d want in a trace, and users who care about where an answer came from will look — which is exactly the trust you’re trying to build by grounding the agent in real sources.