Claude Sonnet 4.5 AI Agent with Composio - Blog Demo

.env.example

# Composio API Configuration
COMPOSIO_API_KEY=your_composio_api_key_here
COMPOSIO_USER_ID=your_email@example.com

# Gmail Authentication
GMAIL_AUTH_CONFIG_ID=your_gmail_auth_config_id

# Twitter Authentication
TWITTER_AUTH_CONFIG_ID=your_twitter_auth_config_id

__init__.py

from .custom_tools import get_custom_tools
from .youtube_transcript import create_youtube_transcript_tool

__all__ = ["get_custom_tools", "create_youtube_transcript_tool"]

agent.py

import os

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

from tools import create_youtube_transcript_tool, get_custom_tools


async def create_agent() -> Agent:
    """
    Create and configure an agent with Composio Tool Router and custom tools.

    Returns:
        Configured Agent instance ready to execute tasks
    """
    composio = Composio(
        api_key=os.getenv("COMPOSIO_API_KEY", ""),
        provider=OpenAIAgentsProvider(),
    )

    # Create Tool Router session with configured toolkits
    session = composio.experimental.tool_router.create_session(
        user_id=os.getenv("COMPOSIO_USER_ID", ""),
        toolkits=[
            {
                "toolkit": "gmail",
                "auth_config_id": os.getenv("GMAIL_AUTH_CONFIG_ID", ""),
            },
            {
                "toolkit": "twitter",
                "auth_config_id": os.getenv("TWITTER_AUTH_CONFIG_ID", ""),
            },
        ],
        manually_manage_connections=True,
    )

    # Gather all tools
    custom_tools = get_custom_tools()
    youtube_tool = create_youtube_transcript_tool()

    agent = Agent(
        name="Assistant",
        instructions="You are a helpful assistant with access to Gmail, Twitter, YouTube transcripts, and custom tools. Be concise and friendly.",
        tools=[
            HostedMCPTool(
                tool_config={
                    "type": "mcp",
                    "server_label": "tool_router",
                    "server_url": session["url"],
                    "require_approval": "never",
                }
            ),
            *custom_tools,
            youtube_tool,
        ],
    )

    return agent

custom_tools.py

import inspect
import json

from agents import FunctionTool


def add_numbers(a: float, b: float):
    """Add two numbers and return the sum."""
    return {"sum": a + b}


async def say_hello(name: str):
    """Return a friendly greeting."""
    return {"message": f"Hello, {name}!"}


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

    Supports both sync and async functions.
    """

    async def on_invoke_tool(ctx, args_json: str):
        args = json.loads(args_json or "{}")
        if inspect.iscoroutinefunction(func):
            result = await func(**args)
        else:
            result = func(**args)
        return json.dumps(result)

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


def get_custom_tools() -> list[FunctionTool]:
    """Return a list of all custom tools."""
    add_numbers_tool = create_function_tool(
        add_numbers,
        name="add_numbers",
        description="Add two numbers and return the sum.",
        schema={
            "type": "object",
            "properties": {
                "a": {"type": "number", "description": "First number"},
                "b": {"type": "number", "description": "Second number"},
            },
            "required": ["a", "b"],
            "additionalProperties": False,
        },
    )

    say_hello_tool = create_function_tool(
        say_hello,
        name="say_hello",
        description="Return a friendly greeting.",
        schema={
            "type": "object",
            "properties": {
                "name": {"type": "string", "description": "Person to greet"},
            },
            "required": ["name"],
            "additionalProperties": False,
        },
    )

    return [add_numbers_tool, say_hello_tool]

main.py

import asyncio

from agents import Runner
from dotenv import load_dotenv
from rich.console import Console
from rich.panel import Panel
from rich.prompt import Prompt
from rich.markdown import Markdown

from agent import create_agent

load_dotenv()

console = Console()


def print_banner():
    """Display welcome banner."""
    banner = """
╔═══════════════════════════════════════════════════╗
║                                                   ║
║     🤖  AI Assistant with Tool Router  🚀        ║
║                                                   ║
╚═══════════════════════════════════════════════════╝
"""
    console.print(banner, style="bold cyan")
    console.print()


def print_examples():
    """Display usage examples."""
    examples = """
**Available Capabilities:**

• **Custom Tools**: Add numbers, generate greetings
• **YouTube**: Fetch video transcripts
• **Gmail**: Manage emails (requires authentication)
• **Twitter**: Interact with Twitter (requires authentication)

**Example Commands:**

• `Add 42 and 58`
• `Say hello to Alice`
• `Get transcript from https://youtube.com/watch?v=...`
• `Fetch contributors from composiohq/composio on GitHub`
• `Check my latest emails`

Type `exit` or `quit` to end the session.
"""
    console.print(Panel(Markdown(examples), border_style="blue", padding=(1, 2)))
    console.print()


async def run_task(agent, task: str):
    """Execute a task with the agent."""
    try:
        console.print(f"[bold yellow]🔄 Processing:[/bold yellow] {task}", style="dim")
        console.print()

        result = await Runner.run(agent, task)

        console.print(
            Panel(
                result.final_output,
                title="[bold green]✨ Response[/bold green]",
                border_style="green",
                padding=(1, 2),
            )
        )
        console.print()

    except Exception as e:
        console.print(
            Panel(
                f"[red]Error: {str(e)}[/red]\n\n"
                "If authentication is required, please complete it in your browser and try again.",
                title="[bold red]⚠️  Task Failed[/bold red]",
                border_style="red",
                padding=(1, 2),
            )
        )
        console.print()


async def main():
    """Main interactive loop."""
    console.clear()
    print_banner()

    with console.status("[bold cyan]Initializing agent...", spinner="dots"):
        agent = await create_agent()

    console.print("[bold green]✓[/bold green] Agent initialized successfully!\n")
    print_examples()

    while True:
        try:
            task = Prompt.ask("[bold cyan]task[/bold cyan]").strip()

            if not task or task.lower() in {"exit", "quit"}:
                console.print("\n[bold cyan]👋 Goodbye![/bold cyan]\n")
                break

            await run_task(agent, task)

        except KeyboardInterrupt:
            console.print("\n\n[bold cyan]👋 Goodbye![/bold cyan]\n")
            break
        except EOFError:
            console.print("\n\n[bold cyan]👋 Goodbye![/bold cyan]\n")
            break


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        console.print("\n[bold cyan]👋 Goodbye![/bold cyan]\n")

youtube_transcript.py

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 YouTube URL."""
    q = urlparse(url)
    host = (q.hostname or "").lower()

    if host == "youtu.be":
        return q.path.lstrip("/") or None
    if host in {"www.youtube.com", "youtube.com", "m.youtube.com"}:
        return parse_qs(q.query).get("v", [None])[0]
    return None


def _fetch_transcript_text(video_id: str, languages: List[str]) -> str:
    """Fetch transcript text using YouTube Transcript API."""
    api = YouTubeTranscriptApi()

    try:
        transcript = api.fetch(video_id, languages=languages)
        return "\n".join(s.text for s in transcript.snippets)

    except NoTranscriptFound:
        transcripts = api.list(video_id)
        for tr in transcripts:
            if any(
                tr.language_code.lower().startswith(code.lower()) for code in languages
            ):
                fetched = tr.fetch()
                return "\n".join(s.text for s in fetched.snippets)
        raise

    except TranscriptsDisabled:
        raise

    except Exception as e:
        raise e


def _yt_transcript_impl(url: str, languages: Optional[List[str]] = None) -> dict:
    """Core implementation for fetching YouTube transcript."""
    langs = languages or ["en", "en-US", "en-GB"]
    video_id = _get_video_id(url)

    if not video_id:
        return {"error": "Invalid YouTube URL. Could not extract a video ID."}

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

        return {
            "video_id": video_id,
            "language_preferences": langs,
            "transcript": text,
        }

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


def create_youtube_transcript_tool() -> FunctionTool:
    """Create a FunctionTool for fetching YouTube transcripts."""

    async def on_invoke_tool(ctx, args_json: str):
        args = json.loads(args_json or "{}")
        url = args.get("url")
        languages = args.get("languages")
        result = _yt_transcript_impl(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.",
        params_json_schema={
            "type": "object",
            "properties": {
                "url": {
                    "type": "string",
                    "description": "YouTube video URL (e.g., https://www.youtube.com/watch?v=dQw4w9WgXcQ)",
                },
                "languages": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Preferred language codes. Default: ['en', 'en-US', 'en-GB']",
                },
            },
            "required": ["url"],
            "additionalProperties": False,
        },
        on_invoke_tool=on_invoke_tool,
        strict_json_schema=True,
    )