Skip to main content
Airweave provides official SDKs for Python and TypeScript/JavaScript, making it easy to integrate semantic search and context retrieval into your AI applications.

Installation

pip install airweave-sdk

Quick Start

from airweave import AirweaveSDK

# Initialize the client
client = AirweaveSDK(
    api_key="YOUR_API_KEY",
    base_url="https://api.airweave.ai"  # Optional: use "http://localhost:8001" for self-hosted
)

# Search a collection
results = client.collections.search(
    readable_id="my-collection",
    query="Find recent failed payments"
)

# Access results
for result in results.results:
    print(f"Score: {result['score']}")
    print(f"Content: {result['payload']['md_content']}")
    print(f"Source: {result['payload']['source_name']}")

Authentication

Both SDKs require an API key for authentication. You can obtain your API key from:
  • Cloud Platform: app.airweave.ai under API Keys
  • Self-Hosted: Your local dashboard at http://localhost:8080
For self-hosted deployments, set base_url="http://localhost:8001" (Python) or baseUrl: 'http://localhost:8001' (TypeScript) when initializing the client.

Core SDK Methods

Collections

Create Collection

collection = client.collections.create(
    name="My First Collection",
    readable_id="my-first-collection",  # Optional: auto-generated if not provided
    description="A collection for customer data"  # Optional
)

print(f"Created collection: {collection.readable_id}")

List Collections

collections = client.collections.list(
    skip=0,
    limit=100
)

for collection in collections:
    print(f"- {collection.name} ({collection.readable_id})")

Get Collection

collection = client.collections.get(
    readable_id="my-collection"
)

print(f"Collection: {collection.name}")
print(f"Created: {collection.created_at}")

Search Collection

Basic search with default parameters:
results = client.collections.search(
    readable_id="my-collection",
    query="customer feedback about pricing"
)

print(f"Found {len(results.results)} results")
Advanced search with all parameters:
from airweave import SearchRequest, Filter, FieldCondition, MatchAny

results = client.collections.search(
    readable_id="my-collection",
    query="customer complaints",
    limit=10,
    offset=0,
    response_type="raw",  # or "completion" for AI-generated answers
    recency_bias=0.7,  # 0.0 to 1.0 - weight recent content higher
    score_threshold=0.6,  # Only return results above this score
    search_method="hybrid",  # "hybrid", "neural", or "keyword"
    expansion_strategy="auto",  # "auto", "llm", or "no_expansion"
    enable_reranking=True,  # LLM-based reranking
    enable_query_interpretation=True  # Extract filters from natural language
)

Delete Collection

client.collections.delete(
    readable_id="my-collection"
)

print("Collection deleted")

Source Connections

Create Source Connection

source_connection = client.source_connections.create(
    name="My Stripe Connection",
    short_name="stripe",  # Available sources: stripe, github, notion, slack, etc.
    readable_collection_id="my-collection",
    authentication={
        "credentials": {
            "api_key": "sk_test_your_stripe_api_key"
        }
    }
)

print(f"Created source connection: {source_connection.id}")
print(f"Status: {source_connection.status}")

List Source Connections

connections = client.source_connections.list(
    skip=0,
    limit=100
)

for connection in connections:
    print(f"- {connection.name} ({connection.short_name}): {connection.status}")

Trigger Sync

job = client.source_connections.run(
    source_connection_id="conn_123456"
)

print(f"Sync started: {job.id}")
print(f"Status: {job.status}")

Get Sync Jobs

jobs = client.source_connections.get_jobs(
    source_connection_id="conn_123456"
)

for job in jobs:
    print(f"Job {job.id}: {job.status} - {job.created_at}")

Search Parameters Reference

query
string
required
The search query text to find relevant documents and data
response_type
string
default:"raw"
Format of the response:
  • raw: Returns search results with scores and metadata
  • completion: Returns an AI-generated natural language answer
limit
integer
default:"100"
Maximum number of results to return (1-1000)
offset
integer
default:"0"
Number of results to skip for pagination
recency_bias
float
How much to weigh recency vs similarity (0.0 to 1.0)
  • 0.0: No recency effect (pure similarity ranking)
  • 1.0: Rank by recency only
  • 0.5: Balance between recency and similarity
score_threshold
float
Minimum similarity score threshold (0.0 to 1.0). Only return results above this score.
search_method
string
default:"hybrid"
Search method to use:
  • hybrid: Combines neural (semantic) + keyword search (recommended)
  • neural: Semantic search only
  • keyword: Text matching only
expansion_strategy
string
default:"auto"
Query expansion strategy:
  • auto: Generates query variations automatically
  • llm: AI-powered query expansion
  • no_expansion: Use exact query without expansion
enable_reranking
boolean
default:"true"
Enable LLM-based reranking to improve result relevance
enable_query_interpretation
boolean
default:"true"
Enable automatic filter extraction from natural language queries

Advanced Examples

AI-Generated Answers (RAG)

Get direct answers instead of raw search results:
results = client.collections.search(
    readable_id="docs-collection",
    query="What is our refund policy?",
    response_type="completion",
    limit=5
)

print(results.completion)  # AI-generated answer based on search results

Recent Content Prioritization

Find the most recent documents about a topic:
results = client.collections.search(
    readable_id="news-collection",
    query="AI breakthroughs",
    recency_bias=0.8,  # Heavily favor recent content
    limit=10
)

High-Quality Results Only

Filter out low-relevance results:
results = client.collections.search(
    readable_id="research-papers",
    query="transformer architectures",
    score_threshold=0.75,  # Only results with 75%+ similarity
    enable_reranking=True
)

Pagination

# Get results 11-20
results = client.collections.search(
    readable_id="my-collection",
    query="customer data",
    limit=10,
    offset=10
)

Error Handling

from airweave import AirweaveSDK
from airweave.exceptions import AirweaveAPIError

try:
    results = client.collections.search(
        readable_id="my-collection",
        query="test query"
    )
except AirweaveAPIError as e:
    print(f"API Error: {e.status_code} - {e.message}")
except Exception as e:
    print(f"Unexpected error: {e}")

SDK Source Code

Python SDK

View the Python SDK source code on GitHub

TypeScript SDK

View the TypeScript SDK source code on GitHub

Need Help?