Integrations Overview

Agent Service
    │
    └─► MCP Client
            │
            ├─► TCP Connection
            │
            └─► MCP Server Process
                    │
                    ├─► Telegram Tools
                    ├─► Gmail Tools
                    ├─► WhatsApp Tools
                    └─► Shopify Tools

services:
  tele_mcp:
    port: 8100
    tools:
      - send_message
      - read_message
      - get_chat_info

  gmail_mcp:
    port: 8101
    tools:
      - send_email
      - read_email
      - list_labels

  whatsapp_mcp:
    port: 8102
    tools:
      - send_message
      - read_message
      - create_group

from Service.McpService import McpService

mcp = McpService()

# Send Telegram message
result = await mcp.execute_tool(
    service="telegram",
    tool="send_message",
    arguments={
        "chat_id": "123456789",
        "text": "Hello from CallIntel!"
    }
)

print(result)

system_prompt = """
You are a customer support agent for TechCorp.

You have access to the following tools:
1. Send email to customers
2. Send Telegram messages
3. Look up orders in Shopify
4. Check email threads

Use these tools to help customers efficiently.
"""

agent = await agent_service.create_agent({
    "name": "Support Agent",
    "system_prompt": system_prompt,
    "available_tools": [
        "gmail.send_email",
        "telegram.send_message",
        "shopify.get_orders"
    ]
})

# When an agent calls a tool
tool_result = await mcp.execute_tool(
    service="shopify",
    tool="get_orders",
    arguments={"customer_id": "cust-123"}
)

# Result is passed back to agent
agent_input = f"Customer orders: {tool_result}"
response = await agent_service.process_message(
    agent_id,
    agent_input
)

TELEGRAM_BOT_TOKEN=your_token
TELEGRAM_API_URL=https://api.telegram.org
TELE_MCP_SERVER_URL=http://tele_mcp:8100

tele_mcp:
  build:
    dockerfile: docker/Dockerfile.tele
  image: tele_mcp_service:latest
  ports:
    - "8100:8100"
  environment:
    - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
    - SUPABASE_URL=${SUPABASE_URL}
    - SUPABASE_KEY=${SUPABASE_KEY}

integrations/mcp_gmail/
├── credentials.json
├── mcp_client.py
├── mcp_server.py
└── predefined_email.py

GMAIL_CREDENTIALS_FILE=integrations/mcp_gmail/credentials.json
GMAIL_MCP_SERVER_URL=http://gmail_mcp:8101

integrations/mcp_whatsapp/
├── database.py
├── mcp_client.py
├── mcp_server.py
└── config.py

WHATSAPP_API_TOKEN=your_token
WHATSAPP_PHONE_NUMBER=your_number
WHATSAPP_MCP_SERVER_URL=http://whatsapp_mcp:8102

integrations/shopify_plugin/
├── client.py
├── models.py
└── auth.py

SHOPIFY_API_KEY=your_key
SHOPIFY_API_SECRET=your_secret
SHOPIFY_STORE_URL=your_store.myshopify.com

# Agent sends email to customer
result = await mcp.execute_tool(
    service="gmail",
    tool="send_email",
    arguments={
        "to": "customer@example.com",
        "subject": f"Order {order_id} Status",
        "body": "Your order has been shipped!"
    }
)

if result["success"]:
    logger.info(f"Email sent to customer")
else:
    logger.error(f"Email send failed: {result['error']}")

# Agent sends message to customer
result = await mcp.execute_tool(
    service="telegram",
    tool="send_message",
    arguments={
        "chat_id": customer_telegram_id,
        "text": "Your appointment is confirmed for tomorrow at 2 PM"
    }
)

# Get response (poll or webhook)
messages = await mcp.execute_tool(
    service="telegram",
    tool="read_message",
    arguments={
        "chat_id": customer_telegram_id,
        "limit": 1
    }
)

agent_response = await agent_service.process_message(
    agent_id,
    f"Customer replied: {messages[0]['text']}"
)

# Agent needs customer data
customer_data = await mcp.execute_tool(
    service="shopify",
    tool="get_customer",
    arguments={
        "customer_id": "cust-123"
    }
)

# Use data in agent prompt
context = f"""
Customer: {customer_data['name']}
Email: {customer_data['email']}
Orders: {customer_data['total_orders']}
Lifetime Value: ${customer_data['total_spent']}
"""

response = await agent_service.process_message(
    agent_id,
    user_message,
    context=context
)

try:
    result = await mcp.execute_tool(
        service="gmail",
        tool="send_email",
        arguments={"to": "user@example.com", "subject": "Hi", "body": "Hello"}
    )
    
    if not result["success"]:
        logger.error(f"Tool failed: {result['error']}")
        # Retry or handle gracefully
        
except ConnectionError:
    logger.error("MCP service unavailable")
    # Fall back to alternative
    
except Exception as e:
    logger.error(f"Unexpected error: {e}")
    # Use fallback mechanism

async def send_message_with_fallback(service, arguments):
    # Try primary service
    try:
        return await mcp.execute_tool(service, "send_message", arguments)
    except Exception as e:
        logger.error(f"{service} failed: {e}")
        
        # Fall back to email
        try:
            return await mcp.execute_tool(
                "gmail",
                "send_email",
                arguments={
                    "to": arguments.get("email"),
                    "subject": "Message",
                    "body": arguments.get("text")
                }
            )
        except:
            # Log and inform user
            logger.error("All message services failed")
            return {"success": False, "error": "Could not send message"}

async def check_integration_health():
    services = ["telegram", "gmail", "whatsapp", "shopify"]
    health = {}
    
    for service in services:
        try:
            result = await mcp.execute_tool(
                service,
                "health_check",
                {}
            )
            health[service] = "healthy" if result["success"] else "unhealthy"
        except:
            health[service] = "unreachable"
    
    return health

async def log_tool_usage(service, tool, arguments, result):
    await supabase.table("integration_logs").insert({
        "service": service,
        "tool": tool,
        "success": result["success"],
        "error": result.get("error"),
        "timestamp": datetime.now(),
        "arguments_count": len(arguments),
    }).execute()

# Check if MCP service is running
curl http://localhost:8100/health

# View service logs
docker-compose logs tele_mcp

# Restart service
docker-compose restart tele_mcp

# Verify credentials are correct
# Check token expiration
# Regenerate tokens if needed

# Implement exponential backoff
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def execute_with_retry(service, tool, arguments):
    return await mcp.execute_tool(service, tool, arguments)