CLI MCP Chat Client with Composio (Developed by Kimi K2 AI Model) - Blog Demo

.env.example

OPENAI_API_KEY=your_openai_api_key_here
COMPOSIO_API_KEY=your_composio_api_key_here

cli_chat.py

#!/usr/bin/env python3
"""
CLI Chat MCP Client with Composio Integration
A chat room interface that connects to MCP servers and handles tool calls via Composio.
"""

import asyncio
import json
import os
import sys
from datetime import datetime
from typing import Any, Dict, List, Optional

import click
from composio import Composio
from dotenv import load_dotenv
from openai import OpenAI
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown
from rich.panel import Panel
from rich.prompt import Prompt
from rich.spinner import Spinner
from rich.table import Table
from rich.text import Text

# Load environment variables
load_dotenv()

console = Console()


class ChatMCPClient:
    def __init__(self):
        self.openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.composio = Composio(api_key=os.getenv("COMPOSIO_API_KEY"))
        self.user_id = None
        self.connected_toolkits = set()
        self.chat_history = []
        self.available_tools = []

    def setup_user(self, user_id: str):
        self.user_id = user_id
        console.print(f"[bold green]Welcome to CLI Chat MCP, {user_id}![/bold green]")

    async def authorize_toolkit(self, toolkit: str) -> bool:
        try:
            console.print(f"[yellow]Authorizing {toolkit}...[/yellow]")
            connection_request = self.composio.toolkits.authorize(
                user_id=self.user_id, toolkit=toolkit.lower()
            )

            console.print(
                Panel(
                    f"🔗 Visit the URL to authorize:\n👉 {connection_request.redirect_url}",
                    title=f"Authorize {toolkit}",
                    border_style="blue",
                )
            )

            with console.status("[bold green]Waiting for authorization..."):
                connection_request.wait_for_connection()

            self.connected_toolkits.add(toolkit)
            console.print(f"[green]✅ {toolkit} authorized successfully![/green]")
            return True

        except Exception as e:
            console.print(f"[red]❌ Failed to authorize {toolkit}: {e}[/red]")
            return False

    def get_available_tools(self, toolkits: List[str] = None) -> List[Dict[str, Any]]:
        try:
            if toolkits:
                tools = self.composio.tools.get(
                    user_id=self.user_id, toolkits=[tk.upper() for tk in toolkits]
                )
            else:
                tools = self.composio.tools.get(user_id=self.user_id)

            self.available_tools = tools
            return tools

        except Exception as e:
            console.print(f"[red]❌ Failed to fetch tools: {e}[/red]")
            return []

    async def process_message(self, message: str) -> str:
        try:
            self.chat_history.append({"role": "user", "content": message})

            messages = [
                {"role": "system", "content": self.get_system_prompt()}
            ] + self.chat_history[-10:]

            completion = self.openai_client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self.available_tools if self.available_tools else None,
                max_tokens=1000,
            )

            response_text = ""

            if completion.choices[0].message.tool_calls:
                console.print("[yellow]🛠️ Executing tool calls...[/yellow]")
                tool_results = self.composio.provider.handle_tool_calls(
                    user_id=self.user_id, response=completion
                )

                for result in tool_results:
                    if isinstance(result, dict) and "status" in result:
                        if result.get("status") == "success":
                            console.print(
                                Panel(
                                    str(
                                        result.get(
                                            "data", "Operation completed successfully"
                                        )
                                    ),
                                    title=f"✅ Tool Result: {result.get('tool_name', 'Unknown')}",
                                    border_style="green",
                                )
                            )
                        else:
                            console.print(
                                Panel(
                                    str(result.get("error", "Operation failed")),
                                    title=f"❌ Tool Error: {result.get('tool_name', 'Unknown')}",
                                    border_style="red",
                                )
                            )

                messages.append(completion.choices[0].message)
                messages.extend(tool_results)

                final_completion = self.openai_client.chat.completions.create(
                    model="gpt-4o", messages=messages, max_tokens=1000
                )

                response_text = final_completion.choices[0].message.content
            else:
                response_text = completion.choices[0].message.content

            self.chat_history.append({"role": "assistant", "content": response_text})
            return response_text

        except Exception as e:
            error_msg = f"Error processing message: {str(e)}"
            console.print(f"[red]{error_msg}[/red]")
            return error_msg

    def get_system_prompt(self) -> str:
        return f"""You are a helpful AI assistant integrated with various tools and services via Composio. 
You can help users with tasks like sending emails, managing calendars, creating documents, and more.
When users ask you to perform actions, use the available tools to complete their requests.
Always provide clear and helpful responses about what actions you've taken.

Available toolkits: {", ".join(self.connected_toolkits) if self.connected_toolkits else "None yet"}
""".strip()

    def display_chat_history(self):
        if not self.chat_history:
            console.print("[dim]No messages yet...[/dim]")
            return

        for msg in self.chat_history:
            if msg["role"] == "user":
                console.print(Panel(msg["content"], title="You", border_style="blue"))
            else:
                console.print(
                    Panel(
                        Markdown(msg["content"]),
                        title="Assistant",
                        border_style="green",
                    )
                )

    def display_help(self):
        help_text = """
# CLI Chat MCP Help

## Commands:
- `/help` - Show this help message
- `/auth <toolkit>` - Authorize a toolkit (e.g., gmail, slack, github)
- `/tools` - List available tools
- `/history` - Show chat history
- `/clear` - Clear chat history
- `/quit` - Exit the chat

## Examples:
- "Send an email to john@example.com with subject 'Meeting' and body 'Let's meet at 3pm'"
- "Create a new GitHub issue in my repo about adding dark mode"
- "Schedule a meeting for tomorrow at 2pm with the team"
- "Get my latest emails from Gmail"

## Available Toolkits:
- gmail - Email management
- slack - Team communication
- github - Repository management
- calendar - Calendar events
- drive - Google Drive files
"""
        console.print(Markdown(help_text))

    def display_tools(self):
        if not self.available_tools:
            console.print(
                "[yellow]No tools available. Try authorizing some toolkits first.[/yellow]"
            )
            return

        table = Table(title="Available Tools")
        table.add_column("Tool", style="cyan")
        table.add_column("Description", style="white")
        table.add_column("Toolkit", style="green")

        for tool in self.available_tools:
            table.add_row(
                tool.get("function", {}).get("name", "Unknown"),
                tool.get("function", {}).get("description", "No description"),
                tool.get("function", {}).get("toolkit", "Unknown"),
            )

        console.print(table)


@click.command()
@click.option("--user-id", default="user", help="User ID for the session")
@click.option("--toolkit", multiple=True, help="Toolkits to authorize on startup")
def main(user_id: str, toolkit: tuple):
    asyncio.run(async_main(user_id, toolkit))


async def async_main(user_id: str, toolkit: tuple):
    if not os.getenv("OPENAI_API_KEY"):
        console.print("[red]❌ OPENAI_API_KEY not found in environment variables[/red]")
        console.print("Please copy .env.example to .env and fill in your API keys")
        return

    if not os.getenv("COMPOSIO_API_KEY"):
        console.print(
            "[red]❌ COMPOSIO_API_KEY not found in environment variables[/red]"
        )
        console.print("Please copy .env.example to .env and fill in your API keys")
        return

    client = ChatMCPClient()
    client.setup_user(user_id)

    if toolkit:
        for tk in toolkit:
            await client.authorize_toolkit(tk)

    client.get_available_tools()

    console.print(
        Panel.fit(
            "[bold cyan]CLI Chat MCP Client[/bold cyan]\n"
            "Type /help for commands, or start chatting!",
            border_style="cyan",
        )
    )

    while True:
        try:
            user_input = Prompt.ask("\n[bold blue]You[/bold blue]")

            if not user_input.strip():
                continue

            if user_input.startswith("/"):
                command = user_input[1:].strip().lower()

                if command in ("quit", "exit"):
                    console.print("[yellow]Goodbye![/yellow]")
                    break
                elif command == "help":
                    client.display_help()
                elif command.startswith("auth "):
                    toolkit_name = command[5:].strip()
                    await client.authorize_toolkit(toolkit_name)
                    client.get_available_tools()
                elif command == "tools":
                    client.display_tools()
                elif command == "history":
                    client.display_chat_history()
                elif command == "clear":
                    client.chat_history.clear()
                    console.print("[green]Chat history cleared![/green]")
                else:
                    console.print(f"[red]Unknown command: {command}[/red]")
                    console.print("Type /help for available commands")
            else:
                with console.status("[bold green]Thinking..."):
                    response = await client.process_message(user_input)

                console.print(
                    Panel(Markdown(response), title="Assistant", border_style="green")
                )

        except KeyboardInterrupt:
            console.print("\n[yellow]Goodbye![/yellow]")
            break
        except Exception as e:
            console.print(f"[red]Error: {e}[/red]")


if __name__ == "__main__":
    main()

requirements.txt

composio
openai
click
rich
prompt-toolkit
asyncio-mqtt
websockets
python-dotenv