Composio Integration

Integrate with Composio platform for seamless third-party service connections

Overview

Kortix provides deep integration with Composio, enabling agents to connect to 150+ third-party services through managed OAuth, MCP servers, and credential profiles. The integration handles authentication, MCP server provisioning, and credential management automatically.

Architecture

The Composio integration consists of several service layers:

┌─────────────────────────────────────────────┐
│     ComposioIntegrationService              │
│  (Orchestrates complete integration flow)   │
└─────────────────────────────────────────────┘
              │
              ├─── ToolkitService
              ├─── AuthConfigService
              ├─── ConnectedAccountService
              ├─── MCPServerService
              └─── ComposioProfileService

Core Services

1. ToolkitService

Module: core/composio_integration/toolkit_service.py

Manages toolkit discovery and information:

• List available toolkits (Gmail, Slack, GitHub, etc.)
• Search toolkits by name or category
• Get toolkit details and required scopes

2. AuthConfigService

Module: core/composio_integration/auth_config_service.py

Handles authentication configuration:

• Create OAuth configs for toolkits
• Manage custom OAuth apps
• Configure scopes and permissions

3. ConnectedAccountService

Module: core/composio_integration/connected_account_service.py

Manages user connections:

• Create connected accounts
• Handle OAuth flow
• Check connection status
• Refresh tokens

4. MCPServerService

Module: core/composio_integration/mcp_server_service.py

Provisions MCP servers:

• Create MCP servers for toolkits
• Generate user-specific MCP URLs
• Configure server permissions

5. ComposioProfileService

Module: core/composio_integration/composio_profile_service.py

Stores credential profiles:

• Save integration configurations
• Retrieve MCP URLs for runtime
• Manage multiple profiles per toolkit

Complete Integration Flow

The ComposioIntegrationService orchestrates the entire integration process:

from core.composio_integration.composio_service import ComposioIntegrationService
from core.services.supabase import DBConnection

# Initialize service
db = DBConnection()
service = ComposioIntegrationService(
    api_key="your_composio_api_key",
    db_connection=db
)

# Complete integration in one call
result = await service.integrate_toolkit(
    toolkit_slug="gmail",
    account_id="user_account_id",
    user_id="user_unique_id",
    profile_name="Gmail for Sales",
    display_name="Gmail Integration",
    save_as_profile=True,
    initiation_fields={"subdomain": "company"}  # If required
)

print(f"MCP URL: {result.final_mcp_url}")
print(f"Profile ID: {result.profile_id}")
print(f"Connected Account: {result.connected_account.id}")

What Happens Behind the Scenes

The integration performs these steps automatically:

async def integrate_toolkit(
    self,
    toolkit_slug: str,
    account_id: str,
    user_id: str,
    profile_name: Optional[str] = None,
    save_as_profile: bool = True,
    initiation_fields: Optional[Dict[str, str]] = None,
    custom_auth_config: Optional[Dict[str, str]] = None,
    use_custom_auth: bool = False
) -> ComposioIntegrationResult:
    # Step 1: Verify toolkit exists
    toolkit = await self.toolkit_service.get_toolkit_by_slug(toolkit_slug)
    
    # Step 2: Create auth config (OAuth)
    auth_config = await self.auth_config_service.create_auth_config(
        toolkit_slug,
        initiation_fields=initiation_fields,
        custom_auth_config=custom_auth_config,
        use_custom_auth=use_custom_auth
    )
    
    # Step 3: Create connected account (user's OAuth connection)
    connected_account = await self.connected_account_service.create_connected_account(
        auth_config_id=auth_config.id,
        user_id=user_id,
        initiation_fields=initiation_fields
    )
    
    # Step 4: Create MCP server
    mcp_server = await self.mcp_server_service.create_mcp_server(
        auth_config_ids=[auth_config.id],
        name=f"{toolkit.name} MCP Server",
        toolkit_name=toolkit.name
    )
    
    # Step 5: Generate user-specific MCP URL
    mcp_url_response = await self.mcp_server_service.generate_mcp_url(
        mcp_server_id=mcp_server.id,
        connected_account_ids=[connected_account.id],
        user_ids=[user_id]
    )
    
    # Prefer account-specific URL (more secure)
    final_mcp_url = mcp_url_response.connected_account_urls[0]
    
    # Step 6: Save as credential profile (optional)
    if save_as_profile:
        profile = await self.profile_service.create_profile(
            account_id=account_id,
            profile_name=profile_name or f"{toolkit.name} Integration",
            toolkit_slug=toolkit_slug,
            mcp_url=final_mcp_url,
            user_id=user_id,
            connected_account_id=connected_account.id
        )
        profile_id = profile.profile_id
    
    return ComposioIntegrationResult(
        toolkit=toolkit,
        auth_config=auth_config,
        connected_account=connected_account,
        mcp_server=mcp_server,
        mcp_url_response=mcp_url_response,
        final_mcp_url=final_mcp_url,
        profile_id=profile_id
    )

Using Composio Tools in Agents

Method 1: Via MCP Configuration

Once integrated, use the MCP URL in your agent configuration:

from core.tools.mcp_tool_wrapper import MCPToolWrapper

# Using the MCP URL from integration
mcp_configs = [
    {
        "name": "gmail_integration",
        "isCustom": True,
        "customType": "composio",
        "config": {
            "profile_id": result.profile_id  # From integration result
        },
        "enabledTools": [  # Optional: filter tools
            "GMAIL_SEND_EMAIL",
            "GMAIL_READ_EMAIL",
            "GMAIL_SEARCH_EMAIL"
        ]
    }
]

mcp_wrapper = MCPToolWrapper(
    mcp_configs=mcp_configs,
    use_cache=True,
    account_id=account_id
)

await mcp_wrapper.initialize_and_register_tools()

Method 2: Direct MCP URL Usage

mcp_configs = [
    {
        "name": "gmail",
        "isCustom": True,
        "customType": "http",
        "config": {
            "url": result.final_mcp_url  # Direct URL from Composio
        }
    }
]

Executing Composio Tools

# Tools are available as methods on the MCP wrapper
result = await mcp_wrapper.GMAIL_SEND_EMAIL(
    to="[e-mail]",
    subject="Hello from Kortix",
    body="This email was sent via Composio integration!"
)

# Or use execute method
result = await mcp_wrapper._execute_mcp_tool(
    tool_name="GMAIL_SEND_EMAIL",
    arguments={
        "to": "[e-mail]",
        "subject": "Hello",
        "body": "Message content"
    }
)

Composio Upload Tool

For Gmail attachments and file operations, use the ComposioUploadTool:

from core.tools.composio_upload_tool import ComposioUploadTool

upload_tool = ComposioUploadTool()

# Upload file to Composio storage
result = await upload_tool.upload_file(
    file_path="/workspace/presentation.pptx",
    file_name="presentation.pptx"
)

# Result contains attachment data
attachment_data = json.loads(result.output)
print(attachment_data)  # {"s3key": "...", "mimetype": "...", "name": "..."}

# Use with Gmail
await mcp_wrapper.GMAIL_SEND_EMAIL(
    to="[e-mail]",
    subject="Presentation",
    body="Please find attached",
    attachment=attachment_data  # Pass attachment data directly
)

Important: When attaching presentations to emails, ALWAYS export and attach .pptx format (not PDF) unless explicitly requested otherwise.

Custom OAuth Apps

For white-label deployments, use custom OAuth credentials:

result = await service.integrate_toolkit(
    toolkit_slug="gmail",
    account_id=account_id,
    user_id=user_id,
    use_custom_auth=True,
    custom_auth_config={
        "client_id": "your_oauth_client_id",
        "client_secret": "your_oauth_client_secret",
        "scopes": ["https://www.googleapis.com/auth/gmail.send"],
        "redirect_uri": "https://your-app.com/oauth/callback"
    }
)

Credential Profiles

Profiles allow users to have multiple integrations for the same toolkit:

from core.composio_integration.composio_profile_service import ComposioProfileService

profile_service = ComposioProfileService(db)

# Create profile
profile = await profile_service.create_profile(
    account_id=account_id,
    profile_name="Work Gmail",
    toolkit_slug="gmail",
    toolkit_name="Gmail",
    mcp_url=mcp_url,
    user_id=user_id,
    is_default=True
)

# List profiles for toolkit
profiles = await profile_service.get_profiles(
    account_id=account_id,
    toolkit_slug="gmail"
)

for profile in profiles:
    print(f"Profile: {profile.profile_name}")
    print(f"Status: {profile.status}")
    print(f"Default: {profile.is_default}")

# Get MCP URL for runtime
mcp_url = await profile_service.get_mcp_url_for_runtime(
    profile_id=profile.profile_id,
    account_id=account_id
)

# Delete profile
await profile_service.delete_profile(
    profile_id=profile.profile_id,
    account_id=account_id
)

Checking Integration Status

# Check if OAuth connection is still valid
status = await service.get_integration_status(
    connected_account_id=result.connected_account.id
)

print(f"Status: {status['status']}")  # "ACTIVE", "EXPIRED", etc.
print(f"Expires at: {status['expires_at']}")

# Refresh if needed
if status['status'] == 'EXPIRED':
    refreshed = await connected_account_service.refresh_account(
        connected_account_id=result.connected_account.id
    )

Available Toolkits

List All Toolkits

toolkits = await service.list_available_toolkits(limit=100)

for toolkit in toolkits['items']:
    print(f"{toolkit['name']} ({toolkit['slug']})")
    print(f"  Category: {toolkit['category']}")
    print(f"  Tools: {toolkit['tool_count']}")

Search Toolkits

results = await service.search_toolkits(
    query="email",
    category="communication",
    limit=20
)

for toolkit in results['items']:
    print(f"Found: {toolkit['name']}")

Popular Toolkits

Communication:

• Gmail - Email operations
• Slack - Team messaging
• Discord - Community chat
• Microsoft Teams - Enterprise messaging

Productivity:

• Google Drive - File storage
• Notion - Knowledge management
• Asana - Project management
• Trello - Task boards

Development:

• GitHub - Code repositories
• GitLab - DevOps platform
• Jira - Issue tracking
• Linear - Project management

CRM & Sales:

• Salesforce - CRM platform
• HubSpot - Marketing & sales
• Zendesk - Customer support
• Intercom - Customer messaging

Calendar & Scheduling:

• Google Calendar - Event management
• Outlook Calendar - Microsoft calendar
• Calendly - Scheduling tool

Environment Variables

Required environment variables:

# Composio API key
COMPOSIO_API_KEY=your_composio_api_key

# Optional: Custom OAuth apps
GMAIL_CLIENT_ID=your_gmail_client_id
GMAIL_CLIENT_SECRET=your_gmail_client_secret

SLACK_CLIENT_ID=your_slack_client_id
SLACK_CLIENT_SECRET=your_slack_client_secret

Error Handling

try:
    result = await service.integrate_toolkit(
        toolkit_slug="gmail",
        account_id=account_id,
        user_id=user_id
    )
except ValueError as e:
    # Toolkit not found or invalid configuration
    logger.error(f"Configuration error: {e}")
except httpx.HTTPError as e:
    # Composio API error
    logger.error(f"API error: {e}")
except Exception as e:
    # Unexpected error
    logger.error(f"Integration failed: {e}", exc_info=True)

Special Toolkit Considerations

Gmail Attachments

Gmail tools have special attachment handling:

# The MCP registry enriches Gmail tool descriptions automatically
def _enrich_description(tool_name: str, description: str) -> str:
    _GMAIL_ATTACHMENT_TOOLS = {
        "GMAIL_SEND_EMAIL", 
        "GMAIL_CREATE_EMAIL_DRAFT", 
        "GMAIL_REPLY_TO_THREAD"
    }
    
    if tool_name.upper() in _GMAIL_ATTACHMENT_TOOLS:
        return description + """
        To attach files: first use composio_upload to upload the file to Composio storage.
        The response includes attachment data (s3key, mimetype, name).
        Pass these to the attachment parameter.
        IMPORTANT: When attaching presentations, always export .pptx (not .pdf) unless requested.
        """
    return description

Zendesk Configuration

Zendesk requires subdomain in initiation_fields:

result = await service.integrate_toolkit(
    toolkit_slug="zendesk",
    account_id=account_id,
    user_id=user_id,
    initiation_fields={
        "subdomain": "yourcompany"  # yourcompany.zendesk.com
    }
)

Best Practices

1. Use Profile Management: Store credentials in profiles for reusability
2. Enable Caching: Use use_cache=True for faster MCP tool loading
3. Filter Tools: Use enabledTools to load only needed tools
4. Handle OAuth Expiry: Check integration status periodically
5. Custom OAuth: Use custom OAuth apps for white-label deployments
6. Error Recovery: Implement retry logic for transient API failures
7. Account-Specific URLs: Prefer connected_account URLs over user_id URLs

Advanced Usage

Multiple Profiles for Same Toolkit

# Work Gmail
work_result = await service.integrate_toolkit(
    toolkit_slug="gmail",
    account_id=account_id,
    user_id="work_user_id",
    profile_name="Work Gmail",
    is_default=True
)

# Personal Gmail
personal_result = await service.integrate_toolkit(
    toolkit_slug="gmail",
    account_id=account_id,
    user_id="personal_user_id",
    profile_name="Personal Gmail",
    is_default=False
)

# Use specific profile
work_mcp_url = await profile_service.get_mcp_url_for_runtime(
    profile_id=work_result.profile_id,
    account_id=account_id
)

Trigger-Based Automations

Composio supports triggers for event-driven workflows:

from core.composio_integration.composio_trigger_service import ComposioTriggerService

trigger_service = ComposioTriggerService()

# Set up Gmail trigger
trigger = await trigger_service.create_trigger(
    toolkit_slug="gmail",
    trigger_name="GMAIL_NEW_EMAIL_RECEIVED",
    connected_account_id=result.connected_account.id,
    webhook_url="https://your-app.com/webhooks/gmail",
    filters={
        "from": "[e-mail]"
    }
)

Next Steps

• MCP Integration - General MCP integration guide
• Custom Tools - Create custom tools
• Built-in Tools - Explore native tools