Context Layer

The Connectors service (client.connectors) 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
Federated search across indexed content
RAG scope and the entity catalog

Connector Types

types = client.connectors.connector_types()
usable = [ct for ct in types if ct.availability is None or ct.availability.available]
 
mysql = client.connectors.connector_type("mysql")
guide = client.connectors.connector_type_setup("aws_s3")

OAuth-class connectors (Google Drive, Slack, Notion, …) cannot complete their browser authorization flow through the SDK. Set them up in the web console — client.connectors.setup_url("google_drive") builds the console deep link, and a headless setup attempt raises ConnectorSetupRequired carrying the same setup_url. Once connected in the console, the connection is fully usable from the SDK.

Create and Manage Connections

conn = client.connectors.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.connectors.connections.list(connector_type="postgresql", page=1, page_size=20)
detail = client.connectors.connections.get(conn.connection_id)
test = client.connectors.connections.test(conn.connection_id)
schema = client.connectors.connections.schema(conn.connection_id, force_refresh=True)
result = client.connectors.connections.query(
    conn.connection_id,
    "SELECT COUNT(*) FROM orders WHERE status = 'active'",
    limit=100,
)

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.test(id)`	`ConnectionTestResult`	Test connection health
`connections.schema(id)`	`SchemaDiscovery`	Discover database schema
`connections.query(id, sql)`	`QueryResult`	Execute a live query (billed at most once per `idempotency_key`)

Sync

config = client.connectors.sync.configure(
    conn.connection_id,
    enabled=True,
    interval_minutes=60,
)
 
trigger = client.connectors.sync.trigger(conn.connection_id)
status = client.connectors.sync.status(conn.connection_id)
jobs = client.connectors.sync.jobs(conn.connection_id, page=1)
status = client.connectors.sync.wait_until_ready(
    conn.connection_id,
    poll_interval=5.0,
    timeout=600.0,
)

RAG Scope

A connection indexes nothing for RAG until you select what to index; the next sync extracts, chunks, and embeds exactly that.

client.connectors.rag.set_scope(conn.connection_id, [
    {"action": "include", "selector": {"kind": "container_id", "value": "folder_123"}},
    {"action": "exclude", "selector": {"kind": "path_prefix", "value": "archive/"}},
])
 
job = client.connectors.rag.sync(conn.connection_id)

RAG Search

results = client.connectors.search(
    "quarterly revenue breakdown by region",
    connection_ids=["conn_abc123"],
    top_k=10,
    rerank=True,
)
for hit in results.results:
    print(hit.citation["entity_name"], hit.match)
 
stats = client.connectors.search_stats()

One call searches the entity catalog (layers=["raw"], free) and the RAG content index (layers=["content"]); the default fuses both. Filters like connector_types, file_types, date_from/date_to, and end_user_id narrow the scope.

Entity Catalog

entities = client.connectors.entities.list(conn.connection_id, page_size=100)
view = client.connectors.entities.content(conn.connection_id, entities.items[0].entity_id)
data = client.connectors.entities.download(conn.connection_id, entities.items[0].entity_id)

Client Lifecycle Multimodal Perception