Integrate Composio into an existing harness

Most Composio examples hand the agent a session and let Composio's meta tools find, authenticate, and run tools. That default works well until you already have your own planner, retrieval index, dispatcher, and permission layer. Then you need the 1000+ apps and managed auth, not another agent inside yours.

Composio sits underneath your harness in four calls:

1. List toolkits. composio.toolkits.get() returns the catalog independent of any user. That's your integrations directory.
2. Connect what the user picks. session.authorize(slug) starts the OAuth flow for this user. Bring your own OAuth credentials via auth configs for a white-label flow, or import existing tokens to migrate from another store.
3. Fetch tools. Load raw JSON Schema with composio.tools.getRawComposioTools({ toolkits }), or serve the same tools over MCP and let your harness call them through the protocol.
4. Execute by slug. session.execute(slug, args) runs one tool as the connected user and hands back data.

Nothing here asks a model to make a decision. Every call is one your code makes, at a moment you choose, because your planner decided.

Setup

You need a Composio API key to list the catalog and drive sessions. Listing apps does not need a user id; creating a session and connecting accounts does. There's no model provider in this example: your harness already has one.

uv add composio

Create a session your code drives

A session is the per-user context: which toolkits are in play, which accounts are connected, what may run. You still want one. What changes is that nothing on it ever reaches a model. Your harness never calls session.tools(), so the search and execute meta tools sit on the session unused.

Two settings are still worth passing at creation. manageConnections: false drops the connection meta-tools, because your settings page is doing that job. Disabling the sandbox drops the code-execution tools, because your harness runs its own code. Leave the toolkit filter open and the session can execute any tool your planner names.

If you want a white-label flow, pass authConfigs at creation to pin the session to your own OAuth app. session.authorize() then sends the user through your consent screen instead of Composio's managed app. You can update those configs later, so migrating from a default app to your own app is a session patch, not a rebuild.

from composio import Composio

composio = Composio()

# user_id is your own identifier, stable for the life of the account
session = composio.sessions.create(
    user_id="user_123",
    manage_connections=False,
    sandbox={"enable": False},
)

Create one session per user and reuse it. Sessions persist on the server and don't expire, so store the session id on the user record and pick it back up with composio.use(sessionId) on later requests. Calling create() again just makes another session.

1. List the toolkits

The catalog does not belong to a user. composio.toolkits.list() in Python or composio.toolkits.get() in TypeScript returns the app list independent of any session, so your integrations page can render before anyone signs in.

page = composio.toolkits.list(limit=50)

for toolkit in page.items:
    print(f"{toolkit.slug:<20} {toolkit.name}")

# page.next_cursor -> pass back as next_cursor= for the next page

Filter by category or sort alphabetically. If you also need to know whether this user has a live connection, call session.toolkits() with the same toolkit list; it merges the global catalog with per-user connection state.

developer = composio.toolkits.list(category="developer-tools", limit=50)
connected = session.toolkits(is_connected=True, limit=100)

2. Connect what the user picks

session.authorize(slug) starts the flow and returns a connection request with a redirect URL. Send the user there. Composio holds the credentials and refreshes them; no token ever lands in your code.

request = session.authorize(
    "github",
    callback_url="https://your-app.com/composio/callback",
)

# Redirect the user here
print(request.redirect_url)

# Scripts and CLIs can block; a web server should not
account = request.wait_for_connection()
print(f"Connected account: {account.id}")

In a web app, don't hold a request open on waitForConnection. Point callbackUrl at your own route and, when the user lands back on it, confirm with session.toolkits({ toolkits: ['github'], isConnected: true }). That check is authoritative and costs one call.

auth_config = composio.auth_configs.create(
    toolkit="github",
    options={
        "type": "use_custom_auth",
        "auth_scheme": "OAUTH2",
        "name": "My GitHub App",
        "credentials": {
            "client_id": os.environ["GITHUB_CLIENT_ID"],
            "client_secret": os.environ["GITHUB_CLIENT_SECRET"],
            "oauth_redirect_uri": "https://backend.composio.dev/api/v1/auth-apps/add",
        },
    },
)

session = composio.sessions.create(
    user_id="user_123",
    toolkits=["github"],
    auth_configs={"github": auth_config.id},
    manage_connections=False,
    sandbox={"enable": False},
)

3. Fetch tools

Your harness can consume tools in two shapes. Load raw JSON Schema and call them yourself, or expose them through an MCP server and let your harness call them over the protocol.

Option A: Raw JSON Schema

getRawComposioTools returns tool definitions with no user context and no provider wrapping: a slug, a name, a description, and input parameters as plain JSON Schema. Take the toolkits you want to expose, pull their tools, and write them into whatever corpus your planner already searches.

connected = session.toolkits(is_connected=True, limit=100)
slugs = [toolkit.slug for toolkit in connected.items]

tools = composio.tools.get_raw_composio_tools(toolkits=slugs)

for tool in tools:
    index.add(
        name=tool.slug,                    # "GITHUB_CREATE_AN_ISSUE"
        description=tool.description,
        parameters=tool.input_parameters,  # JSON Schema
    )

Fetch per toolkit rather than in one shot once a user has a lot connected. Thirty toolkits at full breadth is several thousand schemas, and you almost certainly want them cached, keyed by toolkit and refreshed on your own schedule, rather than pulled on every turn.

Option B: MCP server

Create a session with mcp: true and hand session.mcp.url (and session.mcp.headers if required) to any MCP client. The server exposes the same tools and executes them under the session's connected accounts, so your harness talks the protocol instead of calling session.execute() directly.

from composio import Composio, SESSION_PRESET_DIRECT_TOOLS

composio = Composio()

session = composio.sessions.create(
    user_id="user_123",
    toolkits=["gmail"],
    tools={
        "gmail": {
            "enable": ["GMAIL_FETCH_EMAILS", "GMAIL_CREATE_EMAIL_DRAFT"],
        },
    },
    session_preset=SESSION_PRESET_DIRECT_TOOLS,
    mcp=True,
)

print(session.mcp.url)
print(session.mcp.headers)

Wire the URL into any MCP client. Examples for OpenAI Agents, Claude Agent SDK, and Vercel AI SDK are in the sessions via MCP guide.

Keep the session in step with what you indexed

If you scoped the session to specific toolkits, that filter is what gates execution later. When a user connects something new, widen the session rather than creating another one. session.update() is a patch: fields you omit are left alone.

session.update(toolkits={"enable": [*slugs, "linear"]})

A session created with no toolkit filter, as in the setup above, already allows everything and needs no update.

4. Execute by slug

Your planner picked a tool and produced arguments. session.execute runs it as this user's connected account and returns the result. No model, no retry loop, no interpretation.

result = session.execute(
    "GITHUB_CREATE_AN_ISSUE",
    arguments={
        "owner": "ComposioHQ",
        "repo": "composio",
        "title": "Tool schemas drift between staging and prod",
        "body": "Filed by the harness.",
    },
)

if result.error:
    raise RuntimeError(result.error)

print(result.data)
print(result.log_id)  # look this up in the dashboard

Arguments go through untouched: neither SDK validates them locally, so a wrong field reaches the upstream API and comes back as a populated error your planner can read and correct on the next turn, not as an exception. Exceptions are for transport failures and cancellation. logId ties the call to its entry in the dashboard, which is the fastest way to see what Composio actually sent upstream when a response surprises you.

When the thing you need isn't wrapped as a tool, session.proxyExecute() calls the raw API endpoint under the same connected account.

What maps to what

Your planner, context window, approval gate, and traces stay the same. Composio takes over the global app catalog, OAuth registration, token refresh, schema maintenance, and the execute path, whether you call session.execute() directly or through the MCP server.