Async Client Usage

Async Client Usage

Guide for using the Weaviate Python async client in production applications (FastAPI, async frameworks).

📚 Official Documentation Reference

For agents: If you encounter any issues not covered here, refer to the official Weaviate documentation:

• Primary Reference: Weaviate Async API Documentation
• Python Client Reference: Weaviate Python Client Docs
• API Reference: ReadTheDocs - Python Client
• Troubleshooting: Community Forum | GitHub Issues

Note: The async client (WeaviateAsyncClient) is available in weaviate-client v4.7.0+.

Connection Methods

Three instantiation helpers are provided (docs):

Weaviate Cloud (Recommended)

import weaviate
from weaviate.classes.init import Auth

# Use the official helper function for Weaviate Cloud
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="your-cluster.weaviate.cloud",  # Accepts hostname with or without https://
    auth_credentials=Auth.api_key("your-api-key"),
    headers={  # Note: parameter is "headers" not "additional_headers"
        "X-OpenAI-Api-Key": "sk-...",
        "X-Anthropic-Api-Key": "sk-ant-...",
    }
)

await client.connect()  # Required! Async helpers don't auto-connect

Reference: Weaviate Cloud Setup

Self-Hosted

# For local instances
client = weaviate.use_async_with_local()

# For custom endpoints
client = weaviate.use_async_with_custom(
    http_host="localhost",
    http_port=8080,
    http_secure=False,
    grpc_host="localhost",
    grpc_port=50051,
    grpc_secure=False,
)

await client.connect()

Reference: Connection Configuration

Authentication

Multiple authentication modes are supported (docs):

from weaviate.classes.init import Auth

# API Key (most common for Weaviate Cloud)
auth = Auth.api_key("your-api-key")

# Bearer Token (with optional refresh token)
auth = Auth.bearer_token("access-token", refresh_token="refresh-token")

# Client Credentials (OIDC)
auth = Auth.client_credentials(client_secret="secret")

# Client Password (OIDC Resource Owner Password flow)
auth = Auth.client_password(username="user", password="pass")

# Usage
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="your-cluster.weaviate.cloud",
    auth_credentials=auth,
)

Critical Patterns

⚠️ Connection Lifecycle

Important: Unlike synchronous helpers, async helpers do not connect automatically (docs). You must explicitly call .connect() and .close():

# ❌ Wrong - client not connected
client = weaviate.use_async_with_weaviate_cloud(...)
collections = await client.collections.list_all()  # Will fail!

# ✅ Correct - explicit connect/close
client = weaviate.use_async_with_weaviate_cloud(...)
await client.connect()
collections = await client.collections.list_all()
await client.close()

⚠️ Sync vs Async Methods

Key distinction (docs): Methods involving server requests are async; local operations are synchronous.

# Collection retrieval is SYNC (no await)
collection = client.collections.get("MyCollection")

# Operations on collections are ASYNC (need await)
config = await collection.config.get()
results = await collection.query.fetch_objects()
count = await collection.aggregate.over_all()

Rule: Getting the collection object is sync; calling methods on it is async.

⚠️ Bulk Operations

Important Note (docs): For large-scale data imports, use the synchronous client and its batch operations. The sync client's batch methods already handle concurrency internally and are optimized for bulk operations.

# ✅ For bulk imports, prefer sync client
import weaviate

with weaviate.connect_to_weaviate_cloud(...) as client:
    collection = client.collections.get("MyCollection")

    # Batch insert handles concurrency automatically
    with collection.batch.dynamic() as batch:
        for item in large_dataset:
            batch.add_object(properties=item)

Use the async client for:

• Web applications (FastAPI, Starlette)
• Concurrent request handling
• Interactive queries

Don't use the async client for:

• Bulk data imports (use sync client instead)

Context Manager Pattern (Recommended)

Best Practice (docs): Use async with to automatically connect/disconnect:

from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def get_weaviate_client(
    cluster_url: str,
    api_key: str,
    provider_headers: dict[str, str] | None = None,
) -> AsyncGenerator[weaviate.WeaviateAsyncClient, None]:
    """Connect to Weaviate Cloud with automatic cleanup."""
    # Remove scheme if present
    hostname = cluster_url.replace("https://", "").replace("http://", "")

    client = weaviate.use_async_with_weaviate_cloud(
        cluster_url=hostname,
        auth_credentials=Auth.api_key(api_key),
        headers=provider_headers,
    )

    try:
        await client.connect()
        yield client
    finally:
        await client.close()

# Usage
async def example():
    async with get_weaviate_client(
        cluster_url="your-cluster.weaviate.cloud",
        api_key="your-key",
    ) as client:
        collections = await client.collections.list_all()

Note: When using the context manager, .connect() and .close() are called automatically.

FastAPI Integration

Use Case (docs): The async client excels in web frameworks like FastAPI for handling concurrent requests.

Use lifespan management for shared client across requests:

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: connect to Weaviate
    app.state.weaviate = weaviate.use_async_with_weaviate_cloud(
        cluster_url="your-cluster.weaviate.cloud",
        auth_credentials=Auth.api_key("your-key"),
    )
    await app.state.weaviate.connect()

    yield

    # Shutdown: close connection
    await app.state.weaviate.close()

app = FastAPI(lifespan=lifespan)

@app.get("/collections")
async def list_collections():
    collections = await app.state.weaviate.collections.list_all()
    return {"collections": list(collections.keys())}

Community Discussion: FastAPI Best Practices

Common Pitfalls

1. Parameter Name Confusion

# ❌ Wrong - WeaviateAsyncClient() constructor uses different param
client = weaviate.use_async_with_weaviate_cloud(
    additional_headers={...}  # Wrong parameter name!
)

# ✅ Correct - use "headers" not "additional_headers"
client = weaviate.use_async_with_weaviate_cloud(
    headers={...}
)

2. URL Format

Both formats work with helper functions:

# ✅ Both accepted
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="https://cluster.weaviate.cloud"  # With scheme
)

client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="cluster.weaviate.cloud"  # Without scheme
)

3. Sync vs Async Function Names

# ❌ Wrong - sync client (cannot use await)
client = weaviate.connect_to_weaviate_cloud(...)
await client.connect()  # TypeError!

# ✅ Correct - async client
client = weaviate.use_async_with_weaviate_cloud(...)
await client.connect()

Naming pattern:

• Sync: connect_to_* (e.g., connect_to_weaviate_cloud)
• Async: use_async_with_* (e.g., use_async_with_weaviate_cloud)

4. Port Configuration

# ❌ Wrong - manual port config causes conflicts with Weaviate Cloud
client = WeaviateAsyncClient(
    connection_params=ConnectionParams.from_url(
        url="https://cluster.weaviate.cloud",
        grpc_port=443,  # Conflict!
    )
)

# ✅ Correct - use helper function (handles ports automatically)
client = weaviate.use_async_with_weaviate_cloud(
    cluster_url="cluster.weaviate.cloud"
)

Rule: For Weaviate Cloud, always use use_async_with_weaviate_cloud() — it handles HTTP (443) and gRPC (50051) ports correctly.

Multi-Cluster Example

Managing connections to multiple Weaviate clusters:

@asynccontextmanager
async def get_multi_cluster_clients(
    clusters: dict[str, dict[str, str]]
) -> AsyncGenerator[dict[str, weaviate.WeaviateAsyncClient], None]:
    """Connect to multiple Weaviate clusters.

    Args:
        clusters: Dict of {cluster_id: {"url": "...", "api_key": "..."}}
    """
    clients = {}

    try:
        # Connect to all clusters
        for cluster_id, config in clusters.items():
            client = weaviate.use_async_with_weaviate_cloud(
                cluster_url=config["url"],
                auth_credentials=Auth.api_key(config["api_key"]),
            )
            await client.connect()
            clients[cluster_id] = client

        yield clients

    finally:
        # Close all connections
        for client in clients.values():
            await client.close()

# Usage
async def example():
    clusters = {
        "prod": {"url": "prod.weaviate.cloud", "api_key": "key1"},
        "dev": {"url": "dev.weaviate.cloud", "api_key": "key2"},
    }

    async with get_multi_cluster_clients(clusters) as clients:
        prod_collections = await clients["prod"].collections.list_all()
        dev_collections = await clients["dev"].collections.list_all()

Environment Variables

See Environment Requirements for provider API keys.

import os

# Read from environment
cluster_url = os.environ["WEAVIATE_URL"]
api_key = os.environ["WEAVIATE_API_KEY"]

# Build provider headers
provider_headers = {}
if openai_key := os.getenv("OPENAI_API_KEY"):
    provider_headers["X-OpenAI-Api-Key"] = openai_key
if anthropic_key := os.getenv("ANTHROPIC_API_KEY"):
    provider_headers["X-Anthropic-Api-Key"] = anthropic_key

client = weaviate.use_async_with_weaviate_cloud(
    cluster_url=cluster_url,
    auth_credentials=Auth.api_key(api_key),
    headers=provider_headers or None,
)

Testing Async Code

import pytest

@pytest.mark.asyncio
async def test_weaviate_connection():
    async with get_weaviate_client(
        cluster_url="test-cluster.weaviate.cloud",
        api_key="test-key",
    ) as client:
        collections = await client.collections.list_all()
        assert isinstance(collections, dict)

Quick Reference

Troubleshooting

Common Issues

Getting Help

For agents: When encountering errors:

1. Check the Common Pitfalls section above
2. Search Community Forum for similar issues
3. Check GitHub Issues for known bugs
4. Refer to official async documentation
5. Review Python client best practices

Additional Resources

Official Documentation

• Primary: Weaviate Async API
• Python Client: Main Documentation
• API Reference: ReadTheDocs
• Best Practices: Notes and Best Practices

Framework Integration

• FastAPI Lifespan Events
• Python Async Context Managers

Community

• Weaviate Community Forum
• Python Client GitHub
• Weaviate Blog