Context Layer

Context Layer connects AI applications and agents to external data sources with connector management, sync pipelines, and RAG search.

What it covers

Connector type discovery
Connection lifecycle management
Sync configuration and polling
Search across indexed content
OAuth helpers for supported connectors

Connector Types

types = client.context_hub.connector_types()
for ct in types:
    print(f"{ct.short_name}: {ct.name} — {ct.capabilities}")
 
mysql = client.context_hub.connector_type("mysql")
guide = client.context_hub.connector_type_setup("aws_s3")

Create and Manage Connections

conn = client.context_hub.connections.create(
    name="Production DB",
    connector_type="postgresql",
    credentials={
        "host": "db.example.com",
        "port": 5432,
        "database": "production",
        "username": "readonly",
        "password": "...",
    },
    description="Production read replica",
    labels=["production", "analytics"],
    auto_sync=True,
)
 
page = client.context_hub.connections.list(
    connector_type="postgresql",
    page=1,
    page_size=20,
)
 
detail = client.context_hub.connections.get(conn.connection_id)
client.context_hub.connections.update(conn.connection_id, name="Renamed DB")
test = client.context_hub.connections.test(conn.connection_id)
schema = client.context_hub.connections.schema(conn.connection_id, force_refresh=True)
result = client.context_hub.connections.query(
    conn.connection_id,
    "SELECT COUNT(*) FROM orders WHERE status = 'active'",
    limit=100,
)
objects = client.context_hub.connections.objects(conn.connection_id, path="/")
client.context_hub.connections.delete(conn.connection_id)

Use conn.connection_id for subsequent API calls. Do not use conn.id, which is an internal UUID.

Method	Return Type	Description
`connections.create(...)`	`Connection`	Create a new data connection
`connections.list(...)`	`ConnectionList`	Paginated list of connections
`connections.get(id)`	`ConnectionDetail`	Get connection details
`connections.update(id, ...)`	`Connection`	Update connection properties
`connections.delete(id)`	`None`	Delete a connection
`connections.test(id)`	`ConnectionTestResult`	Test connection health
`connections.schema(id)`	`SchemaDiscovery`	Discover database schema
`connections.query(id, sql)`	`QueryResult`	Execute a live query
`connections.objects(id)`	`dict`	List objects for file connectors

Sync Configuration

config = client.context_hub.sync.get_config(conn.connection_id)
 
config = client.context_hub.sync.configure(
    conn.connection_id,
    enabled=True,
    strategy="full",
    interval_minutes=60,
    include_tables=["orders", "customers"],
    chunk_size=1000,
    chunk_overlap=100,
)
 
client.context_hub.sync.update_config(
    conn.connection_id,
    interval_minutes=30,
)
 
trigger = client.context_hub.sync.trigger(conn.connection_id)
status = client.context_hub.sync.status(conn.connection_id)
jobs = client.context_hub.sync.jobs(conn.connection_id, page=1)
status = client.context_hub.sync.wait_until_ready(
    conn.connection_id,
    poll_interval=5.0,
    timeout=600.0,
)

RAG Search

results = client.context_hub.search(
    "quarterly revenue breakdown by region",
    connection_ids=["conn_abc123"],
    connector_types=["postgresql"],
    top_k=10,
    rerank=True,
    ai_rerank=True,
    expand_query=True,
    file_types=["csv", "pdf"],
    date_from="2025-01-01",
    date_to="2025-12-31",
)
 
stats = client.context_hub.search_stats()

OAuth

url = client.context_hub.oauth.authorize_url(
    "google_drive",
    frontend_redirect="https://app.tou.ai/callback",
)

Client Lifecycle Unstructured Data