Composio Integration

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

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}")

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
    )

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()

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

# Tools are available as methods on the MCP wrapper
result = await mcp_wrapper.GMAIL_SEND_EMAIL(
    to="[email protected]",
    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": "[email protected]",
        "subject": "Hello",
        "body": "Message content"
    }
)

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="[email protected]",
    subject="Presentation",
    body="Please find attached",
    attachment=attachment_data  # Pass attachment data directly
)

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"
    }
)

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
)

# 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
    )

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']}")

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

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

# 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

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)

# 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

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

# 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
)

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": "[email protected]"
    }
)