CLI MCP Chat Client with Composio (Developed by Qwen 3 Coder AI Model) - Blog Demo

composio_mcp_client.py

#!/usr/bin/env python3
"""
CLI Chat MCP Client with Composio Integration

This script provides a command-line interface for chatting and using Composio 
for tool calling and third-party service access based on user messages.
"""

import argparse
import json
import sys
import os
from typing import Dict, Any, List

from composio import Composio
from openai import OpenAI


class ComposioMCPClient:
    def __init__(self, user_id: str):
        # Initialize Composio client
        self.composio = Composio()
        self.openai = OpenAI()  # Will use OPENAI_API_KEY from environment
        self.user_id = user_id
        
        # Check if required environment variables are set
        if not os.getenv("COMPOSIO_API_KEY"):
            raise ValueError("COMPOSIO_API_KEY environment variable is required")
            
        if not os.getenv("OPENAI_API_KEY"):
            raise ValueError("OPENAI_API_KEY environment variable is required")

    def authorize_gmail(self):
        """Authorize Gmail integration for the user."""
        print("Authorizing Gmail integration...")
        try:
            connection_request = self.composio.toolkits.authorize(
                user_id=self.user_id, 
                toolkit="gmail"
            )
            print(f"🔗 Visit the URL to authorize Gmail access:")
            print(f"👉 {connection_request.redirect_url}")
            print("Waiting for authorization to complete...")
            connection_request.wait_for_connection()
            print("✅ Gmail authorization successful!")
        except Exception as e:
            print(f"❌ Gmail authorization failed: {str(e)}")

    def get_composio_tools(self) -> List[Dict[str, Any]]:
        """Fetch available tools from Composio."""
        try:
            tools = self.composio.tools.get(
                user_id=self.user_id, 
                toolkits=["GMAIL"]
            )
            return tools
        except Exception as e:
            print(f"Error fetching tools: {str(e)}")
            return []

    def process_user_message(self, message: str):
        """Process user message and handle tool calls."""
        print(f"\nProcessing: {message}")
        
        # Get available tools
        tools = self.get_composio_tools()
        if not tools:
            print("No tools available. Please authorize Gmail first using --authorize-gmail")
            return
            
        try:
            # Send message to OpenAI with tools
            completion = self.openai.chat.completions.create(
                model="gpt-4o",
                messages=[
                    {
                        "role": "user",
                        "content": message,
                    },
                ],
                tools=tools,
            )
            
            # Handle tool calls
            response = self.composio.provider.handle_tool_calls(
                user_id=self.user_id, 
                response=completion
            )
            
            # Display results
            if response:
                print("\n🔧 Tool Response:")
                print(json.dumps(response, indent=2))
            else:
                # If no tool was called, show the assistant's response
                assistant_message = completion.choices[0].message.content
                if assistant_message:
                    print(f"\n🤖 Assistant: {assistant_message}")
                else:
                    print("\nNo response from AI or tools.")
                    
        except Exception as e:
            print(f"Error processing message: {str(e)}")


def main():
    parser = argparse.ArgumentParser(
        description="CLI Chat MCP Client with Composio Integration",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --user-id user123
  %(prog)s --user-id user123 --authorize-gmail
  %(prog)s --user-id user123 --message "send an email to test@example.com saying hello"
        """
    )
    
    parser.add_argument("--user-id", required=True, help="User ID for the session")
    parser.add_argument("--authorize-gmail", action="store_true", help="Authorize Gmail integration")
    parser.add_argument("--message", help="Single message to process (non-interactive mode)")
    
    args = parser.parse_args()
    
    try:
        # Initialize client
        client = ComposioMCPClient(args.user_id)
        
        # Handle authorization if requested
        if args.authorize_gmail:
            client.authorize_gmail()
            return
            
        # Handle single message mode
        if args.message:
            client.process_user_message(args.message)
            return
            
        # Interactive mode
        print(f"Composio MCP Client - User: {args.user_id}")
        print("Type 'quit' to exit, 'help' for commands.")
        print("-" * 50)
        
        while True:
            try:
                user_input = input("\nYou: ").strip()
                
                if user_input.lower() == 'quit':
                    print("Goodbye!")
                    break
                    
                if user_input.lower() == 'help':
                    print("\nAvailable commands:")
                    print("  quit - Exit the application")
                    print("  help - Show this help message")
                    print("  authorize gmail - Authorize Gmail integration")
                    print("\nExamples:")
                    print("  send an email to test@example.com saying hello")
                    print("  check my recent emails")
                    continue
                    
                if user_input.lower() == 'authorize gmail':
                    client.authorize_gmail()
                    continue
                    
                if not user_input:
                    continue
                    
                # Process user message
                client.process_user_message(user_input)
                
            except KeyboardInterrupt:
                print("\n\nGoodbye!")
                break
            except EOFError:
                print("\n\nGoodbye!")
                break
                
    except Exception as e:
        print(f"Error: {str(e)}")
        sys.exit(1)


if __name__ == "__main__":
    main()

main.py

#!/usr/bin/env python3
"""
CLI Chat MCP Client with Composio Integration

This script provides a command-line interface for chatting with an MCP server
and using Composio for tool calling and third-party service access.
"""

import argparse
import json
import sys
from typing import Dict, Any

from composio import Composio
from openai import OpenAI
import requests


class MCPClient:
    def __init__(self, server_url: str, api_key: str = None):
        self.server_url = server_url.rstrip('/')
        self.api_key = api_key
        self.session = requests.Session()
        if api_key:
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})

    def send_message(self, message: str, user_id: str) -> Dict[str, Any]:
        """Send a message to the MCP server and return the response."""
        payload = {
            "user_id": user_id,
            "message": message
        }
        try:
            response = self.session.post(
                f"{self.server_url}/chat",
                json=payload
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            return {"error": f"Failed to communicate with MCP server: {str(e)}"}


def main():
    parser = argparse.ArgumentParser(description="CLI Chat MCP Client with Composio Integration")
    parser.add_argument("--server", required=True, help="MCP server URL")
    parser.add_argument("--user-id", required=True, help="User ID for the session")
    parser.add_argument("--api-key", help="API key for the MCP server (optional)")
    
    args = parser.parse_args()
    
    # Initialize clients
    mcp_client = MCPClient(args.server, args.api_key)
    openai = OpenAI()  # Will use OPENAI_API_KEY from environment
    composio = Composio()  # Will use COMPOSIO_API_KEY from environment
    
    print(f"Connected to MCP server at {args.server}")
    print("Type 'quit' to exit the chat.")
    print("-" * 40)
    
    while True:
        try:
            user_input = input("\nYou: ").strip()
            if user_input.lower() == 'quit':
                print("Goodbye!")
                break
            
            if not user_input:
                continue
                
            # Send message to MCP server
            response = mcp_client.send_message(user_input, args.user_id)
            
            if "error" in response:
                print(f"Error: {response['error']}")
                continue
                
            # Handle tool calls if present in response
            if response.get("tool_calls"):
                print("\nTool Responses:")
                for tool_call in response["tool_calls"]:
                    tool_name = tool_call.get("name", "unknown")
                    tool_args = tool_call.get("arguments", {})
                    print(f"  - {tool_name}: {json.dumps(tool_args, indent=2)}")
                    
                    # Here you would actually execute the tool call using Composio
                    # For example:
                    # result = composio.execute_tool(tool_name, tool_args, args.user_id)
                    # print(f"    Result: {result}")
            else:
                # Display regular response
                assistant_message = response.get("message", "No response from server")
                print(f"\nAssistant: {assistant_message}")
                
        except KeyboardInterrupt:
            print("\n\nGoodbye!")
            break
        except Exception as e:
            print(f"Error: {str(e)}")
            continue


if __name__ == "__main__":
    main()

requirements.txt

composio
openai