Asynchronous Usage

In modern web applications (like FastAPI or Starlette backends) and high-throughput AI data pipelines, making blocking HTTP requests can severely degrade performance. While waiting for a network response from the Tavily API, your application thread is blocked and unable to handle other tasks.

To solve this, the SDK provides the AsyncTavilyClient, built entirely on the popular asynchronous HTTP library httpx. It fully supports Python's async/await syntax, allowing your event loop to process other operations concurrently.

Basic Initialization & Usage

The AsyncTavilyClient mirrors the exact same API surface as the synchronous client. The only difference is that you must await the network-bound methods.

import asyncio
from tavily import AsyncTavilyClient

async def fetch_info():
    # Initialize the async client
    client = AsyncTavilyClient()

    # Await the API call so the event loop can do other things
    response = await client.search("Current status of the Artemis space program")

    print(f"Found {len(response['results'])} results.")

    # Best practice: Close the client to release connection pools
    await client.close()

# Run the async function
asyncio.run(fetch_info())

High Concurrency Example: Batch Extraction

Where the async client truly shines is in concurrent operations. Suppose you have 50 specific URLs you need to extract content from. Doing this synchronously in a for loop would take a very long time.

Using asyncio.gather, you can fire all requests simultaneously.

import asyncio
from tavily import AsyncTavilyClient

async def batch_extract(url_lists):
    """
    url_lists is a list of lists: 
    [['url1', 'url2'], ['url3', 'url4'], ...]
    """
    async with AsyncTavilyClient() as client:
        # Create a list of concurrent tasks
        tasks = []
        for batch in url_lists:
            # Each extract call can handle up to 20 URLs
            task = client.extract(urls=batch, extract_depth="basic")
            tasks.append(task)

        # Await all tasks concurrently
        # If you have 5 batches, they all execute in parallel
        print(f"Firing {len(tasks)} concurrent API requests...")
        all_results = await asyncio.gather(*tasks)

        return all_results

# Example usage omitted for brevity

Async Streaming (Research API)

When using the Research endpoint with stream=True, the synchronous client returns a standard Python Generator. However, the AsyncTavilyClient returns an AsyncGenerator.

You must use the async for syntax to iterate over the chunks. This is particularly useful if you are streaming the report over WebSockets to a frontend client.

import asyncio
from tavily import AsyncTavilyClient

async def stream_report_to_console():
    async with AsyncTavilyClient() as client:
        # This returns an AsyncGenerator
        stream = await client.research(
            input="History of the UNIX operating system",
            model="mini",
            stream=True
        )

        print("Report Stream Started:\n")

        # Iterating requires 'async for'
        async for chunk in stream:
            # Simulate sending to a websocket or just print
            text = chunk.decode('utf-8')
            print(text, end="", flush=True)

asyncio.run(stream_report_to_console())

Common Pitfalls

Forgetting to Close the Client: The AsyncTavilyClient holds open TCP connections. If you initialize it inside a function and don't call await client.close(), you will leak connections. Always use the async with context manager (as shown in the batch example) to guarantee cleanup.
Mixing Sync and Async: Do not use AsyncTavilyClient inside standard synchronous functions without managing the event loop (e.g., using asyncio.run()), and never use TavilyClient inside an async def function, as it will block the entire event loop.