Composio's default setup assumes one user. Here is how to scale it to hundreds of users each with their own connected accounts.

The Multi-Tenant Challenge

Most Composio tutorials show a single user authenticating a single app. In a real SaaS product, you have hundreds or thousands of users, each with their own connected accounts (their own Slack, their own Gmail, their own GitHub). Every agent action must use that specific user's credentials -- never another user's.

Composio handles this through Entities. This article explains the Entity pattern and the production setup required to make it safe and scalable.

The Entity Pattern

In Composio, an Entity represents a user in your system. Each Entity has its own connected accounts. When you run an agent action, you specify the Entity, and Composio uses that Entity's credentials automatically.

from composio import Composio
 
client = Composio(api_key="your-composio-api-key")
 
# Create an entity for a user (do this when the user signs up)
# entity_id should be your app's user ID -- any stable string
entity = client.get_entity(id="user-123")  # creates if not exists
 
# Initiate the connection flow for this entity
connection_request = entity.initiate_connection(
    app_name="slack",
    redirect_url="https://yourapp.com/auth/callback?userId=user-123",
)
 
# Send this URL to your frontend for the user to authorise
print(f"Authorisation URL: {connection_request.redirectUrl}")

Using Entity Credentials in Agent Workflows

from composio_langchain import ComposioToolSet
 
# Key: create a toolset scoped to a specific entity
def get_tools_for_user(user_id: str):
    toolset = ComposioToolSet(
        api_key="your-composio-api-key",
        entity_id=user_id,   # all tool calls use this user's credentials
    )
    return toolset.get_tools(actions=[
        "SLACK_SEND_MESSAGE",
        "GMAIL_SEND_EMAIL",
        "GITHUB_CREATE_ISSUE",
    ])
 
# In your request handler:
async def handle_agent_request(user_id: str, user_message: str):
    tools = get_tools_for_user(user_id)  # scoped to this specific user
 
    agent = create_react_agent(llm, tools)
    result = await agent.ainvoke({"input": user_message})
    return result
Never share a toolset across users. Always instantiate a new ComposioToolSet with the specific user's entity_id for each request. Sharing toolsets can result in one user's agent using another user's credentials.

Checking What Apps a User Has Connected

from composio import Composio
 
client = Composio(api_key="your-api-key")
 
def get_user_connections(user_id: str) -> list:
    entity = client.get_entity(id=user_id)
    connections = entity.get_connections()
 
    return [
        {
            "app": conn.appName,
            "status": conn.status,
            "connected_at": conn.createdAt,
            "account_id": conn.id,
        }
        for conn in connections
    ]
 
# Use this to show users what they have connected in your settings UI
connections = get_user_connections("user-123")
for conn in connections:
    print(f"{conn['app']}: {conn['status']}")

Gating Agent Actions by Connected Apps

Before running an agent that needs a specific integration, verify the user has connected it. This prevents confusing errors and gives you the opportunity to prompt the user to connect the app first.

async def run_slack_agent(user_id: str, message: str) -> str:
    client = Composio(api_key="your-api-key")
    entity = client.get_entity(id=user_id)
 
    # Check if Slack is connected before running
    connections = {c.appName: c.status for c in entity.get_connections()}
 
    if "slack" not in connections:
        return "Please connect your Slack account first in Settings > Integrations."
 
    if connections["slack"] != "ACTIVE":
        return "Your Slack connection has expired. Please reconnect in Settings > Integrations."
 
    # Safe to run -- user has an active Slack connection
    tools = get_tools_for_user(user_id)
    agent = create_react_agent(llm, tools)
    result = await agent.ainvoke({"input": message})
    return result["output"]

Handling Connection Expiry at Scale

OAuth tokens expire and sometimes get revoked by users on the third-party app's side. At scale, you will have users with expired connections running agent tasks. Build a proactive check into your scheduled jobs or agent middleware.

# Scheduled job: scan for expired connections and notify users
async def audit_expired_connections():
    client = Composio(api_key="your-api-key")
 
    # Fetch all entities (paginated in practice)
    all_users = await db.get_all_user_ids()
 
    for user_id in all_users:
        entity = client.get_entity(id=user_id)
        connections = entity.get_connections()
 
        for conn in connections:
            if conn.status in ("EXPIRED", "FAILED"):
                await notify_user_to_reconnect(
                    user_id=user_id,
                    app_name=conn.appName,
                )

Quick Reference

  • Use entity_id=user_id when creating ComposioToolSet -- always scope to the specific user
  • Never share a toolset across users -- one toolset per request, scoped to that user's entity
  • Check connection status before running agent actions -- return a clear reconnect prompt if expired
  • Create entities proactively at user sign-up -- do not wait for the first agent action
  • Run a scheduled audit job for expired connections and notify users proactively
  • Store the account_id returned from initiate_connection in your database for direct reference