SDK ReferencePython SDK

ConnectedAccounts

Methods

refresh()

Refresh the credentials of a connected account.

def refresh(nanoid: str, **params: Unpack[connected_account_refresh_params.ConnectedAccountRefreshParams]) -> connected_account_refresh_response.ConnectedAccountRefres...

Parameters

NameTypeDescription
nanoidstrThe connected account ID (ca_xxx).
params?Unpack[connected_account_refresh_params.ConnectedAccountRefreshParams]

Returns

connected_account_refresh_response.ConnectedAccountRefres... — The refresh response.

Example

composio.connected_accounts.refresh("ca_abc123")

revoke()

Revoke a connected account's upstream tokens (best effort) and mark the connection REVOKED. Raises composio.exceptions.ComposioConnectedAccountRevocationNotSupportedError when the toolkit does not support programmatic revocation (API 400) and composio.exceptions.ComposioConnectedAccountNotRevokableError when the connection is not in a revokable state (API 409).

def revoke(nanoid: str) -> connected_account_revoke_response.ConnectedAccountRevokeR...

Parameters

NameTypeDescription
nanoidstrThe connected account ID (ca_xxx).

Returns

connected_account_revoke_response.ConnectedAccountRevokeR... — The revoked tokens and the updated connected account.

Example

result = composio.connected_accounts.revoke("ca_abc123")
print(result.connected_account.status)  # "REVOKED"

complete_auth()

Complete a deferred OAuth connection once you have verified the user's identity. When your project has an OAuth callback verifier configured, Composio does not activate a new OAuth connection by itself: it redirects to your verifier with a single-use session URI. After confirming who the user is, redeem it here; Composio checks that your project owns the pending connection and that user_id is its owner, completes the token exchange, and the connection becomes ACTIVE. Redeeming a session twice, or after it expires, fails with a 404 from the API.

def complete_auth(*, user_id: str, session_uri: str) -> connected_account_complete_auth_response.ConnectedAccount...

Parameters

NameTypeDescription
user_idstrThe user the connection was initiated for.
session_uristrThe session URI from the verifier redirect.

Returns

connected_account_complete_auth_response.ConnectedAccount... — The completed connected_account_id and its toolkit_slug.

Example

result = composio.connected_accounts.complete_auth(
    user_id="user_123",
    session_uri=session_uri,  # from the verifier redirect
)
print(result.connected_account_id)

update()

Update a connected account's alias and/or credentials.

def update(nanoid: str, *, alias: str | None = ..., connection: connected_account_patch_params.Connection | None = ...) -> connected_account_patch_response.ConnectedAccountPatchRes...

Parameters

NameTypeDescription
nanoidstrThe connected account ID (ca_xxx).
alias?str | NoneHuman-readable alias. Pass an empty string to clear. Must be unique per entity and toolkit within the project.
connection?connected_account_patch_params.Connection | NoneCredential update with authScheme and val fields.

Returns

connected_account_patch_response.ConnectedAccountPatchRes... — Response with id, status, and success.

Example

# Set an alias
composio.connected_accounts.update('ca_abc123', alias='work-gmail')

# Clear an alias
composio.connected_accounts.update('ca_abc123', alias='')

update_acl()

Update the per-user ACL on a SHARED connected account. Experimental — shape may change in future releases. Only valid on SHARED connections; raises ComposioAclOnlyForSharedError on a PRIVATE connection. Omit a parameter to leave it unchanged; pass an empty list to clear an allow/deny list. At least one parameter must be provided.

def update_acl(nanoid: str, *, allow_all_users: bool | None = ..., allowed_user_ids: List[str] | None = ..., not_allowed_user_ids: List[str] | None = ...) -> connected_account_patch_response.ConnectedAccountPatchRes...

Parameters

NameTypeDescription
nanoidstrThe connected account ID (ca_xxx).
allow_all_users?bool | NoneWhen True, any user_id may use this SHARED connection (subject to the deny list).
allowed_user_ids?List[str] | NoneExplicit list of allowed user_id strings. Pass [] to clear.
not_allowed_user_ids?List[str] | NoneExplicit deny list (wins over allow on conflict). Pass [] to clear — note that clearing the deny list silently re-grants access to previously-blocked users.

Returns

connected_account_patch_response.ConnectedAccountPatchRes... — Response with id, status, and success.

Example

composio.connected_accounts.update_acl(
    'ca_abc',
    allow_all_users=True,
    not_allowed_user_ids=['user_bob'],
)

initiate()

Compound function to create a new connected account. This function creates a new connected account and returns a connection request. Users can then wait for the connection to be established using the wait_for_connection method.

def initiate(user_id: str, auth_config_id: str, *, callback_url: str | None = ..., allow_multiple: bool = ..., config: connected_account_create_params.ConnectionState | None = ..., alias: str | None = ...) -> ConnectionRequest

Parameters

NameTypeDescription
user_idstrThe user ID to create the connected account for.
auth_config_idstrThe auth config ID to create the connected account for.
callback_url?str | NoneCallback URL to use for OAuth apps.
allow_multiple?boolWhether to allow multiple connected accounts for the same user and auth config.
config?connected_account_create_params.ConnectionState | NoneThe configuration to create the connected account with.
alias?str | NoneOptional human-readable alias for the account. Must be unique per userId and toolkit within the project.

Returns

ConnectionRequest — The connection request.


Create a Composio Connect Link for a user to connect their account to a given auth config. This method will return an external link which you can use for the user to connect their account.

def link(user_id: str, auth_config_id: str, *, callback_url: str | None = ..., alias: str | None = ..., allow_multiple: bool = ..., experimental: link_create_params.Experimental | None = ...) -> ConnectionRequest

Parameters

NameTypeDescription
user_idstrThe external user ID to create the connected account for.
auth_config_idstrThe auth config ID to create the connected account for.
callback_url?str | NoneThe URL to redirect the user to post connecting their account.
alias?str | NoneOptional human-readable alias for the connection. Must be unique per userId and toolkit within the project.
allow_multiple?boolWhether to allow multiple connected accounts for the same user and auth config. When False (default), raises ComposioMultipleConnectedAccountsError if the user already has an ACTIVE connection on this auth config. Pair with alias and a session-level multi_account config to disambiguate at execution time.
experimental?link_create_params.Experimental | NoneExperimental options for this connection. Pass an Experimental dict with account_type and/or acl_config_for_shared to create a SHARED connection with a per-user ACL. Experimental — shape may change in future releases.

Returns

ConnectionRequest — Connection request object.

Example

# Create a connection request and redirect the user to the redirect url
connection_request = composio.connected_accounts.link('user_123', 'auth_config_123')
redirect_url = connection_request.redirect_url
print(f"Visit: {redirect_url} to authenticate your account")

# Wait for the connection to be established
connected_account = connection_request.wait_for_connection()

# Create a connection request with callback URL
connection_request = composio.connected_accounts.link(
    'user_123',
    'auth_config_123',
    callback_url='https://your-app.com/callback'
)
redirect_url = connection_request.redirect_url
print(f"Visit: {redirect_url} to authenticate your account")

# Wait for the connection to be established
connected_account = composio.connected_accounts.wait_for_connection(connection_request.id)

connection_request = composio.connected_accounts.link(
    'user_creator',
    'auth_config_123',
    experimental={
        'account_type': 'SHARED',
        'acl_config_for_shared': {
            'allow_all_users': True,
            'not_allowed_user_ids': ['user_bob'],
        },
    },
)

wait_for_connection()

Wait for connected account with given ID to be active

def wait_for_connection(id: str, timeout: float | None = ...) -> connected_account_retrieve_response.ConnectedAccountRetri...

Parameters

NameType
idstr
timeout?float | None

Returns

connected_account_retrieve_response.ConnectedAccountRetri...


View source