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 )
The user ID to create the connected account for.
The auth config ID to use for authentication.
Callback URL for OAuth redirects.
Allow multiple connected accounts for the same user and auth config.
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
The connected account ID.
Connection status (e.g., “INITIATED”, “ACTIVE”).
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