Skip to main content

​
What You’ll Learn

  • Submitting a research request and polling for results
  • Streaming research progress in real time
  • Handling streaming events (progress updates, source discoveries, content generation)
  • Getting structured output with custom schemas

​
How Does It Work?

Tavily Research is a multi-agent deep research endpoint. You submit a query, and Tavily’s research agents autonomously search, extract, and synthesize information from multiple sources into a comprehensive report. Two consumption patterns are available:
PatternBest ForHow It Works
PollingBackground jobs, serverlessSubmit request, poll for completion
StreamingReal-time UIs, CLI toolsReceive progress events as they happen

​
Getting Started

Get your Tavily API key

uv venv
uv pip install tavily-python
# Optional (used in the production helper example below)
uv pip install tavily-agent-toolkit

​
Polling

client.research() returns immediately with a pending task. Use client.get_research() to poll until the status is "completed". 
import os
import time
from tavily import TavilyClient

client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

task = client.research(
    input="What are the key trends in AI agents for 2026?",
    model="mini",
)

print(f"Task created: {task['request_id']} (status: {task['status']})")

while True:
    result = client.get_research(task["request_id"])

    if result["status"] == "completed":
        break
    elif result["status"] == "failed":
        raise RuntimeError("Research task failed")

    print(f"Status: {result['status']}...")
    time.sleep(5)

print(result["content"])
print(f"\nSources: {len(result['sources'])}")
for source in result["sources"]:
    print(f"  - {source['title']}: {source['url']}")

​
Streaming

Stream research progress in real time. The API sends Server-Sent Events (SSE) in an OpenAI-compatible chat.completion.chunk format as the research agents work. You can display tool activity, show discovered sources, and stream the final report as it generates. 
import json
import os
from tavily import TavilyClient

client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

stream = client.research(
    input="What are the key trends in AI agents for 2026?",
    model="mini",
    stream=True,
)

for chunk in stream:
    text = chunk.decode("utf-8")

    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue

        if line.startswith("event:"):
            event_name = line.split(":", 1)[1].strip()
            if event_name == "done":
                print("\n\n[stream complete]")
            continue

        if not line.startswith("data:"):
            continue

        payload = line.split(":", 1)[1].strip()
        if not payload:
            continue

        data = json.loads(payload)
        delta = data.get("choices", [{}])[0].get("delta", {})

        if "tool_calls" in delta:
            tc = delta["tool_calls"]
            if tc.get("type") == "tool_call":
                for tool in tc.get("tool_call", []):
                    print(f"[{tool.get('name')}] {tool.get('arguments', '')}")
            elif tc.get("type") == "tool_response":
                for tr in tc.get("tool_response", []):
                    for s in tr.get("sources", []):
                        print(f"  Source: {s['title']} — {s['url']}")

        if "content" in delta and isinstance(delta["content"], str):
            print(delta["content"], end="", flush=True)

        if "sources" in delta:
            print("\n\n--- All Sources ---")
            for s in delta["sources"]:
                print(f"  {s['title']}: {s['url']}")

​
Streaming Event Flow

A typical session progresses through these stages:
StageDelta fieldWhat happens
Planningtool_calls (type tool_call)Research plan is initialized
Searchingtool_calls (type tool_call / tool_response)Web searches execute, sources are discovered
Subtopicstool_calls (Pro only)ResearchSubtopic calls for deeper dives
Generatingtool_callsFinal report generation begins
ContentcontentReport chunks stream incrementally
SourcessourcesComplete list of all sources used
DoneSSE event: doneStream complete
For full event structure and field-level details, see the Streaming API reference.

​
Structured Output

Use output_schema to get research results in a custom JSON format. Define your schema as a dictionary of field names to descriptions. 
import os
import json
import time
from tavily import TavilyClient

client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

task = client.research(
    input="Compare React, Vue, and Svelte for building web apps",
    model="mini",
    output_schema={
        "properties": {
            "frameworks": {
                "type": "array",
                "description": "List of frameworks compared",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string", "description": "Framework name"},
                        "strengths": {
                            "type": "array",
                            "description": "Key strengths",
                            "items": {"type": "string"}
                        },
                        "weaknesses": {
                            "type": "array",
                            "description": "Key weaknesses",
                            "items": {"type": "string"}
                        },
                        "best_for": {"type": "string", "description": "Best use case"}
                    },
                    "required": ["name", "best_for"]
                }
            },
            "recommendation": {"type": "string", "description": "Overall recommendation"}
        },
        "required": ["frameworks", "recommendation"]
    },
)

while True:
    result = client.get_research(task["request_id"])
    if result["status"] == "completed":
        break
    if result["status"] == "failed":
        raise RuntimeError("Research task failed")
    time.sleep(5)

content = result["content"]
structured = json.loads(content) if isinstance(content, str) else content

for fw in structured["frameworks"]:
    print(f"\n{fw['name']}: Best for {fw['best_for']}")

​
Handling Research Streams in Production

For production applications, use the handle_research_stream utility from the Agent Toolkit to process streaming events cleanly: 
import os
from tavily import TavilyClient
from tavily_agent_toolkit import handle_research_stream

client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])

response = client.research(
    input="Analysis of the AI search market landscape",
    model="pro",
    stream=True,
)

report = handle_research_stream(
    response,
    stream_content_generation=True,
)

print(report)

​
Critical Knobs

  • "auto" (default) — Tavily picks based on query complexity
  • "pro" — comprehensive, multi-agent research for complex topics
  • "mini" — faster, targeted research for narrow questions
  • true — real-time SSE progress events
  • false (default) — single response, poll for completion
  • Pass a JSON Schema object to get structured data instead of a markdown report
  • Write clear field descriptions for best results
  • "numbered" (default), "mla", "apa", "chicago"
  • Controls how sources are cited in the report
For the complete parameter list, see the Research API reference.

​
Next Steps

Research API Reference

Full parameter list, response schema, and interactive playground.

Streaming API Reference

Complete SSE event structure, field details, and tool types.

Research Best Practices

Prompting tips, model selection, and structured output guidance.

Python SDK Reference

Python client methods, async support, and type details.