Connected Accounts

Manage third-party service connections and authentication

Connected Accounts represent authenticated connections to third-party services. The ConnectedAccounts class provides methods to create, manage, and maintain these connections.

Accessing Connected Accounts

from composio import Composio

composio = Composio()
accounts = composio.connected_accounts

Methods

initiate

Create a new connected account and return a connection request.

connection_request = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_xxx",
    callback_url="https://your-app.com/callback"
)

print(f"Redirect URL: {connection_request.redirect_url}")

# Wait for connection
connected_account = connection_request.wait_for_connection(timeout=60)

user_id · str · required

The user ID to create the connected account for.

auth_config_id · str · required

The auth config ID to use for authentication.

callback_url · str

Callback URL for OAuth redirects.

allow_multiple · bool · default: "False"

Allow multiple connected accounts for the same user and auth config.

config · ConnectionState

Connection configuration (for API key, basic auth, etc.).

link

Create a Composio Connect Link for user authentication.

connection_request = composio.connected_accounts.link(
    user_id="user_123",
    auth_config_id="ac_xxx",
    callback_url="https://your-app.com/callback"
)

print(f"Visit: {connection_request.redirect_url}")

get

Retrieve a specific connected account.

account = composio.connected_accounts.get("ca_xxx")
print(f"Status: {account.status}")

list

List connected accounts with filtering.

accounts = composio.connected_accounts.list(
    user_ids=["user_123"],
    auth_config_ids=["ac_xxx"],
    statuses=["ACTIVE"]
)

for account in accounts.items:
    print(f"{account.id}: {account.status}")

enable

Enable a connected account.

composio.connected_accounts.enable("ca_xxx")

disable

Disable a connected account.

composio.connected_accounts.disable("ca_xxx")

refresh

Refresh account credentials.

composio.connected_accounts.refresh("ca_xxx")

delete

Delete a connected account.

composio.connected_accounts.delete("ca_xxx")

wait_for_connection

Wait for a connection to become active.

account = composio.connected_accounts.wait_for_connection(
    id="ca_xxx",
    timeout=60
)

Connection Request

The ConnectionRequest object is returned by initiate() and link().

Properties

id · str

The connected account ID.

status · str

Connection status (e.g., “INITIATED”, “ACTIVE”).

redirect_url · str | None

OAuth redirect URL for user authorization.

Methods

wait_for_connection · (timeout: float) -> ConnectedAccount

Wait for the connection to become active.

account = connection_request.wait_for_connection(timeout=300)

Auth Schemes

Use the auth_scheme helper to create connections with different authentication methods.

from composio import auth_scheme

# API Key
connection = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_xxx",
    config=auth_scheme.api_key(
        options={"api_key": "your-api-key"}
    )
)

# Basic Auth
connection = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_xxx",
    config=auth_scheme.basic(
        options={
            "username": "user",
            "password": "pass"
        }
    )
)

# Bearer Token
connection = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_xxx",
    config=auth_scheme.bearer_token(
        options={"token": "your-token"}
    )
)

Connection Statuses

• ACTIVE: Connection is established and working
• INACTIVE: Connection is temporarily disabled
• PENDING: Connection is being processed
• INITIATED: Connection request has started
• EXPIRED: Connection credentials have expired
• FAILED: Connection attempt failed

Examples

OAuth Flow

# Step 1: Initiate connection
connection = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_github",
    callback_url="https://myapp.com/callback"
)

# Step 2: Redirect user
print(f"Please visit: {connection.redirect_url}")

# Step 3: Wait for completion
try:
    account = connection.wait_for_connection(timeout=300)
    print(f"Success! Account ID: {account.id}")
except Exception as e:
    print(f"Connection failed: {e}")

API Key Authentication

from composio import auth_scheme

connection = composio.connected_accounts.initiate(
    user_id="user_123",
    auth_config_id="ac_xxx",
    config=auth_scheme.api_key(
        options={"api_key": "sk_live_xxx"}
    )
)

account = connection.wait_for_connection()
print(f"Connected: {account.id}")

List User Connections

accounts = composio.connected_accounts.list(
    user_ids=["user_123"],
    statuses=["ACTIVE"]
)

for account in accounts.items:
    print(f"Toolkit: {account.toolkit.name}")
    print(f"Status: {account.status}")
    print(f"Created: {account.created_at}")

Refresh Expired Connection

try:
    composio.connected_accounts.refresh("ca_xxx")
    print("Credentials refreshed successfully")
except Exception as e:
    print(f"Refresh failed: {e}")

Next Steps

Auth Configs

Configure authentication

Tools

Execute tools with connections

Toolkits

Authorize toolkits