Instances & Credentials¶

A connector descriptor defines what a connector can do. An instance binds that descriptor to specific credentials and configuration for a particular environment or use case. This separation lets you run the same connector type (e.g., "PostgreSQL") with different credentials for different databases, teams, or environments.

Connector Instances¶

Creating an Instance¶

To use a connector, create an instance tied to a specific connector descriptor:

from flow_sdk import FlowSDK

sdk = FlowSDK()

instance = sdk.connector_instances.create(
    connector_id="uuid-of-postgres-connector",
    instance_name="Analytics DB - Production",
    description="Read-only access to the production analytics database",
    configuration={
        "host": "analytics-db.internal.company.com",
        "port": 5432,
        "database": "analytics",
        "ssl_mode": "require",
    },
)

print(f"Instance created: {instance.id}")

Field	Required	Description
`connector_id`	Yes	UUID of the connector descriptor
`instance_name`	Yes	Human-readable name (1-255 chars)
`description`	No	What this instance is for
`workspace_id`	No	Scope to a specific workspace (null = org-level)
`configuration`	No	Non-sensitive settings (host, port, database, etc.)
`auth_type`	No	Which authentication profile to use

Configuration vs credentials

The configuration field stores non-sensitive settings like hostnames and ports. Sensitive values (passwords, tokens, API keys) go in credentials, which are encrypted at rest and stored separately.

Instance Lifecycle¶

Instances track their health and connection status:

Field	Description
`is_enabled`	Whether the instance is active (can be disabled without deletion)
`last_tested_at`	When the connection was last tested
`test_status`	`success`, `failed`, or `pending`
`test_message`	Human-readable test result detail
`has_credentials`	Whether credentials have been stored

Updating an Instance¶

sdk.connector_instances.update(
    instance_id="uuid-of-instance",
    instance_name="Analytics DB - Production (Read-Only)",
    configuration={
        "host": "analytics-db-v2.internal.company.com",
        "port": 5432,
        "database": "analytics",
        "ssl_mode": "require",
    },
)

Disabling an Instance¶

Disable an instance without deleting it. Disabled instances cannot be used by agents or workflows.

sdk.connector_instances.update(
    instance_id="uuid-of-instance",
    is_enabled=False,
)

Credential Management¶

Credentials are stored separately from instance configuration. They are encrypted before storage and never returned in full through the API.

Credential Scopes¶

Credentials are scoped to control who can use them:

Scope	Description	Use Case
`organization`	Available to all users in the org	Shared service accounts
`workspace`	Available within a specific workspace	Team-specific credentials
`user`	Personal credentials for one user	Individual API tokens
`agent`	Dedicated credentials for a specific agent	Isolated agent access

Storing Credentials¶

credential = sdk.connector_instances.create_credential(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
    credentials={
        "username": "readonly_user",
        "password": "s3cur3_p4ssw0rd",
    },
    credential_metadata={
        "label": "Production read-only user",
        "created_for": "analytics pipeline",
    },
    expires_at="2025-12-31T23:59:59Z",  # optional
)

print(f"Credential stored: {credential.id}")

Credentials are write-once

The API returns only credential metadata after creation -- never the plaintext values. To update credentials, create a new credential record. The platform uses the most recent valid credential for each scope.

Credential Expiration¶

For OAuth tokens and time-limited credentials, set expires_at. The platform tracks expiration and can alert you when credentials are about to expire:

expiring = sdk.connector_instances.list_expiring_credentials(
    days_ahead=30,
)

for cred in expiring:
    print(f"{cred.connector_name}: expires in {cred.days_remaining:.0f} days")

OAuth2 Flows¶

For connectors using OAuth2, the platform handles the token exchange:

Authorization Code flow: The platform generates an authorization URL. The user completes consent in the browser. The platform exchanges the authorization code for access and refresh tokens.
Client Credentials flow: The platform exchanges client ID and secret for an access token automatically.

Refresh tokens are handled transparently -- the platform refreshes expired access tokens before making API calls.

Connection Testing¶

After storing credentials, test the connection to verify everything works:

result = sdk.connector_instances.test_connection(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
)

print(f"Status: {result.status}")     # success or failed
print(f"Message: {result.message}")
print(f"Tested at: {result.tested_at}")

The test connection call uses the connector's test_endpoint configuration (for API key / basic auth) or performs a token exchange (for OAuth2) to verify credentials are valid.

Health Checks¶

For ongoing monitoring, run lightweight health checks:

health = sdk.connector_instances.health_check(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
    timeout_seconds=10.0,
)

print(f"Status: {health.status}")      # healthy or unhealthy
print(f"Latency: {health.latency_ms}ms")

Bulk Operations¶

Run health checks or credential refreshes across multiple instances at once:

results = sdk.connector_instances.bulk_operation(
    instance_ids=["uuid-1", "uuid-2", "uuid-3"],
    action="health_check",
)

for r in results:
    print(f"{r.instance_id}: {r.status}")

Maximum batch size is 50 instances per request.

Schema Discovery¶

For connectors with supports_schema_discovery: true, retrieve the schema of available tables and fields:

schema = sdk.connector_instances.get_schema(
    instance_id="uuid-of-instance",
    scope_type="organization",
    scope_id="uuid-of-org",
)

for table in schema.tables:
    print(f"{table.name}: {[f.name for f in table.fields]}")

DDL Execution (Database Connectors)¶

For database connectors, execute DDL statements against the connected database. Use confirm=False to preview the operation before committing.

# Preview (dry run)
result = sdk.connector_instances.execute_ddl(
    instance_id="uuid-of-instance",
    sql="CREATE TABLE events (id SERIAL PRIMARY KEY, name TEXT, created_at TIMESTAMPTZ)",
    confirm=False,
    scope_type="workspace",
    scope_id="uuid-of-workspace",
)
print(result.preview)

# Execute for real
result = sdk.connector_instances.execute_ddl(
    instance_id="uuid-of-instance",
    sql="CREATE TABLE events (id SERIAL PRIMARY KEY, name TEXT, created_at TIMESTAMPTZ)",
    confirm=True,
    scope_type="workspace",
    scope_id="uuid-of-workspace",
    reason="Creating events table for analytics pipeline",
)

Parameter	Required	Description
`instance_id`	Yes	Connector instance UUID
`sql`	Yes	SQL statement(s) to execute
`confirm`	Yes	`False` = preview only; `True` = execute
`scope_type`	Yes	`workspace` or `organization`
`scope_id`	No	Defaults to the configured workspace ID
`reason`	No	Required for destructive operations (DROP, TRUNCATE) on non-empty tables

Destructive operations

DDL statements that would affect non-empty tables (e.g., DROP TABLE, TRUNCATE) require a reason to be provided. The platform logs all DDL executions for audit purposes.

Instance Architecture¶

graph TD
    CI["Connector Instance<br/>instance_name, configuration"] --> CD["Connector Descriptor<br/>operations, auth profiles"]
    CI --> Creds["Credentials<br/>(encrypted at rest)"]

    Creds --> OrgScope["Org Scope"]
    Creds --> WSScope["Workspace Scope"]
    Creds --> UserScope["User Scope"]
    Creds --> AgentScope["Agent Scope"]

    CI --> Agent["Agent<br/>(via MCP tools)"]
    CI --> Workflow["Workflow<br/>(via operation calls)"]
    CI --> DatasetSrc["Dataset Source<br/>(via query API)"]

Best Practices¶

One instance per environment -- Create separate instances for dev, staging, and production.
Use organization scope for shared credentials -- Avoid duplicating credentials across users.
Set credential expiration -- Even for long-lived API keys, set an expiration to force periodic rotation.
Test after credential updates -- Always run a connection test after storing new credentials.
Disable before deleting -- Disable an instance first to verify nothing breaks, then delete.