Cursor Composer 1 AI Agent with Composio - Blog Demo

__init__.py

"""
Tools Package

This package contains all custom tools and tool creation utilities.
"""

from tools.custom_tools import add_numbers, say_hello
from tools.tool_factory import create_function_tool
from tools.youtube_tool import create_youtube_transcript_tool

__all__ = [
    "add_numbers",
    "say_hello",
    "create_function_tool",
    "create_youtube_transcript_tool",
]

agent_builder.py

"""
Agent Builder

Creates and configures the AI agent with Composio Tool Router and custom tools.
"""

import os

from agents import Agent, HostedMCPTool
from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider
from dotenv import load_dotenv

from tools import create_youtube_transcript_tool

# Load environment variables
load_dotenv()


def get_toolkits_config() -> list:
    """
    Build toolkits configuration from environment variables.

    Reads auth config IDs from environment:
    - GMAIL_AUTH_CONFIG_ID: Auth config ID for Gmail toolkit
    - TWITTER_AUTH_CONFIG_ID: Auth config ID for Twitter toolkit

    Only includes toolkits that have their auth_config_id set in environment.

    Returns:
        List of toolkit configuration dictionaries
    """
    toolkits = []

    gmail_auth_id = os.getenv("GMAIL_AUTH_CONFIG_ID")
    if gmail_auth_id:
        toolkits.append({"toolkit": "gmail", "auth_config_id": gmail_auth_id})

    twitter_auth_id = os.getenv("TWITTER_AUTH_CONFIG_ID")
    if twitter_auth_id:
        toolkits.append({"toolkit": "twitter", "auth_config_id": twitter_auth_id})

    return toolkits


def create_custom_tools():
    """
    Create all custom FunctionTool instances.

    Returns:
        List of FunctionTool instances
    """
    youtube_tool = create_youtube_transcript_tool()

    return [youtube_tool]


async def build_agent() -> Agent:
    """
    Build and configure the AI agent with all available tools.

    Sets up:
    - Composio Tool Router (Gmail, Twitter) - configured via environment variables
    - Custom local tools (add_numbers, say_hello, yt_transcript)

    Environment Variables Required:
    - COMPOSIO_API_KEY: Your Composio API key
    - COMPOSIO_USER_ID: Your user ID/email (defaults to empty string if not set)

    Optional Environment Variables:
    - GMAIL_AUTH_CONFIG_ID: Auth config ID for Gmail toolkit
    - TWITTER_AUTH_CONFIG_ID: Auth config ID for Twitter toolkit

    Returns:
        Configured Agent instance ready to use
    """
    # Initialize Composio with OpenAI provider
    composio_api_key = os.getenv("COMPOSIO_API_KEY") or ""
    if not composio_api_key:
        raise ValueError(
            "COMPOSIO_API_KEY environment variable is required. "
            "Please set it in your .env file or environment."
        )

    composio = Composio(
        api_key=composio_api_key,
        provider=OpenAIAgentsProvider(),
    )

    # Get user ID from environment (defaults to empty string)
    user_id = os.getenv("COMPOSIO_USER_ID", "")

    # Build toolkits configuration from environment variables
    toolkits = get_toolkits_config()

    if not toolkits:
        print(
            "⚠️  Warning: No toolkits configured. "
            "Set GMAIL_AUTH_CONFIG_ID and/or TWITTER_AUTH_CONFIG_ID in your .env file."
        )

    # Create Tool Router session with configured toolkits
    session = composio.experimental.tool_router.create_session(
        user_id=user_id,
        toolkits=toolkits,
        manually_manage_connections=True,
    )

    # Create all custom tools
    custom_tools = create_custom_tools()

    # Build the agent with Tool Router and custom tools
    agent = Agent(
        name="Assistant",
        instructions=(
            "You are a helpful AI assistant with access to various tools. "
            "You can use the Composio Tool Router to interact with Gmail and Twitter, "
            "as well as custom local tools for calculations, greetings, and YouTube transcripts. "
            "Always be helpful, clear, and concise in your responses."
        ),
        tools=[
            # Composio Tool Router (MCP server)
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "tool_router",
                    "server_url": session["url"],
                    "require_approval": "never",
                }
            ),
            # Custom local tools
            *custom_tools,
        ],
    )

    return agent

cli.py

"""
Command Line Interface

Beautiful, user-friendly CLI for interacting with the AI agent.
"""

import sys
from typing import Optional

from agents import Agent, Runner


class Colors:
    """ANSI color codes for terminal output."""

    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"

    # Text colors
    BLUE = "\033[34m"
    CYAN = "\033[36m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    RED = "\033[31m"
    MAGENTA = "\033[35m"

    # Background colors
    BG_BLUE = "\033[44m"
    BG_CYAN = "\033[46m"
    BG_GREEN = "\033[42m"


def print_header():
    """Print a beautiful header banner."""
    header = f"""
{Colors.BOLD}{Colors.CYAN}{"=" * 70}
{Colors.RESET}{Colors.BOLD}{Colors.CYAN}  🤖 AI Assistant - Powered by Composio Tool Router
{Colors.RESET}{Colors.CYAN}{"=" * 70}{Colors.RESET}
"""
    print(header)


def print_welcome():
    """Print welcome message with examples."""
    examples = [
        ("Get transcript from https://youtube.com/watch?v=...", "YouTube transcript"),
        ("Send an email", "Gmail integration"),
        ("Post a tweet", "Twitter integration"),
    ]

    print(f"{Colors.BOLD}{Colors.GREEN}Welcome!{Colors.RESET}")
    print(
        f"{Colors.DIM}Type your task below or 'exit'/'quit' to leave.{Colors.RESET}\n"
    )

    print(f"{Colors.BOLD}Example tasks:{Colors.RESET}")
    for task, category in examples:
        print(
            f"  {Colors.CYAN}•{Colors.RESET} {Colors.DIM}{task:<45}{Colors.RESET} {Colors.YELLOW}({category}){Colors.RESET}"
        )
    print()


def print_separator(char: str = "─", length: int = 70):
    """Print a horizontal separator line."""
    print(f"{Colors.DIM}{char * length}{Colors.RESET}")


def print_task_prompt():
    """Print the task input prompt."""
    prompt = (
        f"{Colors.BOLD}{Colors.BLUE}💬 You{Colors.RESET} {Colors.DIM}>{Colors.RESET} "
    )
    return prompt


def print_thinking():
    """Print a thinking indicator."""
    print(f"\n{Colors.DIM}{Colors.YELLOW}🤔 Thinking...{Colors.RESET}\n")


def print_result(result: str):
    """Print the agent's result in a formatted box."""
    print(f"\n{Colors.BOLD}{Colors.GREEN}{'═' * 70}")
    print(f"  ✨ Assistant Response")
    print(f"{'═' * 70}{Colors.RESET}\n")
    print(f"{result}\n")
    print(f"{Colors.DIM}{'─' * 70}{Colors.RESET}\n")


def print_error(error: Exception):
    """Print error message in a formatted way."""
    print(f"\n{Colors.BOLD}{Colors.RED}{'═' * 70}")
    print(f"  ⚠️  Error")
    print(f"{'═' * 70}{Colors.RESET}\n")
    print(f"{Colors.RED}{str(error)}{Colors.RESET}\n")

    # Check if it's an authentication error
    error_str = str(error).lower()
    if "auth" in error_str or "authenticate" in error_str:
        print(
            f"{Colors.YELLOW}💡 Tip:{Colors.RESET} If you were prompted to authenticate, "
        )
        print(f"   complete the authentication in your browser and try again.\n")

    print(f"{Colors.DIM}{'─' * 70}{Colors.RESET}\n")


def print_goodbye():
    """Print a friendly goodbye message."""
    print(f"\n{Colors.BOLD}{Colors.CYAN}{'=' * 70}")
    print(f"  👋 Thanks for using AI Assistant! Have a great day!")
    print(f"{'=' * 70}{Colors.RESET}\n")


async def run_task(agent: Agent, task: str) -> Optional[str]:
    """
    Execute a task with the agent and display results.

    Args:
        agent: The configured agent instance
        task: User's task description

    Returns:
        Final output string if successful, None otherwise
    """
    print_thinking()

    try:
        result = await Runner.run(agent, task)
        print_result(result.final_output)
        return result.final_output
    except KeyboardInterrupt:
        print(f"\n{Colors.YELLOW}⚠️  Task interrupted by user.{Colors.RESET}\n")
        return None
    except Exception as e:
        print_error(e)
        return None


async def run_interactive_session(agent: Agent):
    """
    Run the main interactive CLI session.

    Args:
        agent: The configured agent instance
    """
    print_header()
    print_welcome()

    while True:
        try:
            # Get user input
            task = input(print_task_prompt()).strip()

            # Check for exit commands
            if not task or task.lower() in {"exit", "quit", "q"}:
                break

            # Run the task
            await run_task(agent, task)

        except KeyboardInterrupt:
            print(
                f"\n{Colors.YELLOW}⚠️  Interrupted. Type 'exit' to quit.{Colors.RESET}\n"
            )
        except EOFError:
            # Handle Ctrl+D
            print()
            break

    print_goodbye()

main.py

"""
Main Entry Point

AI Assistant CLI application with Composio Tool Router integration.
"""

import asyncio
import sys

from agent_builder import build_agent
from cli import run_interactive_session


async def main():
    """
    Main application entry point.
    
    Initializes the agent and starts the interactive CLI session.
    """
    # Build the agent with all configured tools
    agent = await build_agent()
    
    # Start interactive session
    await run_interactive_session(agent)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 Goodbye!")
        sys.exit(0)

tool_factory.py

"""
Tool Factory

Helper utilities for creating FunctionTool instances from Python functions.
"""

import inspect
import json
from typing import Callable, Dict

from agents import FunctionTool


def create_function_tool(
    func: Callable,
    name: str,
    description: str,
    schema: Dict,
) -> FunctionTool:
    """
    Convert a Python function into a FunctionTool.

    Supports both synchronous and asynchronous functions.
    Automatically handles JSON serialization of results.

    Args:
        func: Python function to wrap (can be sync or async)
        name: Tool name (used by the agent)
        description: Human-readable description of what the tool does
        schema: JSON schema defining the tool's parameters

    Returns:
        Configured FunctionTool instance
    """

    async def on_invoke_tool(ctx, args_json: str):
        """Handler that invokes the wrapped function."""
        args = json.loads(args_json or "{}")

        # Handle both sync and async functions
        if inspect.iscoroutinefunction(func):
            result = await func(**args)
        else:
            result = func(**args)

        # Return JSON string for consistency
        return json.dumps(result)

    return FunctionTool(
        name=name,
        description=description,
        params_json_schema=schema,
        on_invoke_tool=on_invoke_tool,
        strict_json_schema=True,
    )

youtube_tool.py

"""
YouTube Transcript Tool

This module provides functionality to fetch transcripts from YouTube videos.
Supports multiple YouTube URL formats and language preferences.
"""

import json
from typing import List, Optional
from urllib.parse import parse_qs, urlparse

from agents import FunctionTool
from youtube_transcript_api import (
    NoTranscriptFound,
    TranscriptsDisabled,
    YouTubeTranscriptApi,
)


def _get_video_id(url: str) -> Optional[str]:
    """
    Extract video ID from various YouTube URL formats.
    
    Supports:
    - https://www.youtube.com/watch?v=VIDEO_ID
    - https://youtu.be/VIDEO_ID
    - https://m.youtube.com/watch?v=VIDEO_ID
    
    Args:
        url: YouTube video URL
        
    Returns:
        Video ID if found, None otherwise
    """
    parsed = urlparse(url)
    hostname = (parsed.hostname or "").lower()

    # Handle youtu.be short URLs
    if hostname == "youtu.be":
        return parsed.path.lstrip("/") or None
    
    # Handle standard YouTube URLs
    if hostname in {"www.youtube.com", "youtube.com", "m.youtube.com"}:
        video_id = parse_qs(parsed.query).get("v", [None])[0]
        return video_id
    
    return None


def _fetch_transcript_text(video_id: str, languages: List[str]) -> str:
    """
    Fetch transcript text for a YouTube video.
    
    Uses the YouTube Transcript API to retrieve transcripts in the preferred
    languages. Falls back to available transcripts if preferred ones aren't found.
    
    Args:
        video_id: YouTube video ID
        languages: List of preferred language codes (e.g., ['en', 'en-US'])
        
    Returns:
        Transcript text as a single string
        
    Raises:
        NoTranscriptFound: If no transcript is available
        TranscriptsDisabled: If transcripts are disabled for the video
    """
    api = YouTubeTranscriptApi()

    try:
        # Try to fetch transcript in preferred languages
        transcript = api.fetch(video_id, languages=languages)
        return "\n".join(snippet.text for snippet in transcript.snippets)

    except NoTranscriptFound:
        # Fallback: list available transcripts and pick a matching one
        available_transcripts = api.list(video_id)
        for transcript_info in available_transcripts:
            # Check if language code matches any of our preferences
            if any(
                transcript_info.language_code.lower().startswith(lang.lower())
                for lang in languages
            ):
                fetched = transcript_info.fetch()
                return "\n".join(snippet.text for snippet in fetched.snippets)
        raise


def _fetch_youtube_transcript(url: str, languages: Optional[List[str]] = None) -> dict:
    """
    Main function to fetch YouTube transcript.
    
    Args:
        url: YouTube video URL
        languages: Optional list of preferred language codes
        
    Returns:
        Dictionary with transcript data or error information
    """
    default_languages = ["en", "en-US", "en-GB"]
    preferred_languages = languages or default_languages
    
    video_id = _get_video_id(url)
    if not video_id:
        return {"error": "Invalid YouTube URL. Could not extract video ID."}

    try:
        transcript_text = _fetch_transcript_text(video_id, preferred_languages)
        if not transcript_text.strip():
            return {"error": "Transcript is empty or not available."}

        return {
            "video_id": video_id,
            "language_preferences": preferred_languages,
            "transcript": transcript_text,
        }

    except TranscriptsDisabled:
        return {"error": "Transcripts are disabled for this video."}
    except NoTranscriptFound:
        return {
            "error": f"No transcript found for languages {preferred_languages}."
        }
    except Exception as e:
        return {
            "error": f"Failed to fetch transcript. {type(e).__name__}: {str(e)}"
        }


def create_youtube_transcript_tool() -> FunctionTool:
    """
    Create a FunctionTool for fetching YouTube video transcripts.
    
    Returns:
        Configured FunctionTool instance
    """
    async def on_invoke_tool(ctx, args_json: str):
        """Handler for tool invocation."""
        args = json.loads(args_json or "{}")
        url = args.get("url")
        languages = args.get("languages")
        result = _fetch_youtube_transcript(url=url, languages=languages)
        return json.dumps(result)

    return FunctionTool(
        name="yt_transcript",
        description=(
            "Fetch and return the transcript text for a YouTube video URL. "
            "Accepts optional preferred languages. Useful for summarizing videos, "
            "extracting information, or analyzing video content."
        ),
        params_json_schema={
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": (
                        "YouTube video URL. Examples: "
                        "https://www.youtube.com/watch?v=dQw4w9WgXcQ or "
                        "https://youtu.be/dQw4w9WgXcQ"
                    ),
                },
                "languages": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": (
                        "Preferred language codes (e.g., ['en', 'es', 'fr']). "
                        "Defaults to ['en', 'en-US', 'en-GB'] if not specified."
                    ),
                },
            },
            "required": ["url"],
            "additionalProperties": False,
        },
        on_invoke_tool=on_invoke_tool,
        strict_json_schema=True,
    )