Usage Examples

This document provides practical examples of using the Collaboration Tools MCP Server in various scenarios.

Table of Contents

1. Web Scraping with Notifications
2. Scheduled Health Checks
3. Admin Approval Workflow
4. Multi-Channel Alerting
5. Browser Automation Pipeline

---

Web Scraping with Notifications

Monitor a website and send alerts when specific content appears.

async def monitor_for_keyword(agent, url, keyword, check_interval=3600):
    """Check website for keyword and alert if found."""

    # Set up recurring check
    timer = await agent.call_tool("mcp_set_recurring_timer", {
        "interval_seconds": check_interval,
        "timer_name": f"Monitor {keyword} on {url}",
        "callback_message": f"Check {url} for {keyword}"
    })

    # Initial check
    await agent.call_tool("mcp_browser_navigate", {"url": url})
    content = await agent.call_tool("mcp_browser_get_content", {})

    if keyword in content["content"]:
        # Keyword found! Alert via multiple channels
        await agent.call_tool("mcp_send_email", {
            "to_email": "team@example.com",
            "subject": f"🔍 Keyword '{keyword}' found on {url}",
            "body": f"The keyword '{keyword}' was detected on {url}"
        })

        await agent.call_tool("mcp_send_slack_message", {
            "message": f"🎯 Found '{keyword}' on {url}!"
        })

        # Take screenshot as evidence
        await agent.call_tool("mcp_browser_screenshot", {
            "full_page": True
        })

---

Scheduled Health Checks

Perform regular health checks with escalation.

async def health_check_workflow(agent, service_url):
    """Monitor service health and escalate issues."""

    # Check every 5 minutes
    await agent.call_tool("mcp_set_recurring_timer", {
        "interval_seconds": 300,
        "timer_name": "Health Check",
        "callback_message": "Perform health check"
    })

    # Navigate to health endpoint
    result = await agent.call_tool("mcp_browser_navigate", {
        "url": f"{service_url}/health"
    })

    if not result["success"]:
        # Service down - escalate to admin
        approval = await agent.call_tool("mcp_request_admin_approval", {
            "request_message": f"Service {service_url} is down. Restart service?",
            "context": {"service": service_url, "error": result["error"]},
            "timeout_seconds": 300,
            "urgent": True
        })

        if approval["approved"]:
            # Admin approved restart
            await agent.call_tool("mcp_send_telegram_message", {
                "message": f"🔧 Restarting {service_url}..."
            })
            # ... perform restart ...
        else:
            # Notify team of ongoing issue
            await agent.call_tool("mcp_send_email", {
                "to_email": "oncall@example.com",
                "subject": f"🚨 Service Down: {service_url}",
                "body": "Service is down and restart was not approved."
            })

---

Admin Approval Workflow

Request human approval for sensitive operations.

async def database_maintenance(agent):
    """Perform database maintenance with admin approval."""

    # Step 1: Analyze database
    print("Analyzing database...")
    # ... analysis code ...
    records_to_delete = 50000

    # Step 2: Request approval
    approval = await agent.call_tool("mcp_request_admin_approval", {
        "request_message": f"Delete {records_to_delete} old records from database?",
        "context": {
            "operation": "delete",
            "table": "logs",
            "count": records_to_delete,
            "estimated_time": "5 minutes"
        },
        "timeout_seconds": 600,
        "urgent": False
    })

    if not approval["approved"]:
        print("❌ Operation cancelled by admin")
        return

    # Step 3: Perform deletion with progress updates
    await agent.call_tool("mcp_send_slack_message", {
        "message": f"🗑️ Starting deletion of {records_to_delete} records..."
    })

    # Set timer to check progress
    await agent.call_tool("mcp_set_timer", {
        "duration_seconds": 300,
        "timer_name": "Deletion timeout",
        "callback_message": "Check if deletion completed"
    })

    # ... perform deletion ...

    # Step 4: Notify completion
    await agent.call_tool("mcp_send_email", {
        "to_email": approval["admin_email"],
        "subject": "✅ Database Maintenance Complete",
        "body": f"Successfully deleted {records_to_delete} records.\n\n"
                f"Notes: {approval['admin_notes']}"
    })

---

Multi-Channel Alerting

Send alerts across multiple communication channels.

async def critical_alert(agent, title, message, severity="high"):
    """Send critical alert via all available channels."""

    emoji = "🚨" if severity == "high" else "⚠️"
    full_message = f"{emoji} {title}\n\n{message}"

    # Send to all channels in parallel
    tasks = []

    # Email
    tasks.append(agent.call_tool("mcp_send_email", {
        "to_email": "alerts@example.com",
        "subject": f"{emoji} {title}",
        "body": message,
        "cc": ["oncall@example.com"]
    }))

    # Slack
    tasks.append(agent.call_tool("mcp_send_slack_message", {
        "message": full_message,
        "channel": "#alerts"
    }))

    # Telegram
    tasks.append(agent.call_tool("mcp_send_telegram_message", {
        "message": full_message,
        "parse_mode": None
    }))

    # Discord
    tasks.append(agent.call_tool("mcp_send_discord_message", {
        "message": full_message
    }))

    # Wait for all to complete
    results = await asyncio.gather(*tasks)

    success_count = sum(1 for r in results if r.get("success"))
    print(f"Alert sent via {success_count}/{len(tasks)} channels")

    # If high severity and email/Slack failed, request admin intervention
    if severity == "high" and success_count < 2:
        await agent.call_tool("mcp_request_admin_approval", {
            "request_message": "Alert delivery partially failed. Manual notification needed?",
            "context": {"title": title, "channels_failed": len(tasks) - success_count},
            "urgent": True
        })

---

Browser Automation Pipeline

Complex multi-step browser automation workflow.

async def competitor_research(agent, competitor_url):
    """Research competitor and compile report."""

    print("🔍 Starting competitor research...")

    # Step 1: Navigate and take initial screenshot
    await agent.call_tool("mcp_browser_navigate", {
        "url": competitor_url
    })

    screenshot1 = await agent.call_tool("mcp_browser_screenshot", {
        "full_page": True
    })

    # Step 2: Extract pricing information
    print("📊 Extracting pricing...")
    pricing_result = await agent.call_tool("mcp_browser_execute_task", {
        "task": f"Go to {competitor_url} and extract all pricing plans with their features",
        "max_steps": 30
    })

    # Step 3: Check their blog for recent posts
    print("📝 Checking blog...")
    await agent.call_tool("mcp_browser_execute_task", {
        "task": "Find the blog and extract titles of the 5 most recent posts",
        "max_steps": 20
    })

    blog_screenshot = await agent.call_tool("mcp_browser_screenshot", {
        "full_page": False
    })

    # Step 4: Request admin review of findings
    print("👤 Requesting admin review...")
    review = await agent.call_tool("mcp_request_admin_input", {
        "prompt": "Review competitor research findings. Any additional areas to investigate?",
        "input_type": "text",
        "timeout_seconds": 7200  # 2 hours
    })

    # Step 5: If admin provided additional areas, research them
    if review["success"] and review["input"]:
        print(f"🔍 Investigating additional area: {review['input']}")
        await agent.call_tool("mcp_browser_execute_task", {
            "task": f"Research: {review['input']}",
            "max_steps": 25
        })

    # Step 6: Compile and send report
    print("📧 Sending report...")
    await agent.call_tool("mcp_send_email", {
        "to_email": "team@example.com",
        "subject": f"Competitor Research: {competitor_url}",
        "body": f"""
Competitor Research Report

URL: {competitor_url}
Screenshots: {screenshot1['path']}, {blog_screenshot['path']}

Pricing Info:
{pricing_result['result']}

Admin Notes:
{review.get('input', 'None')}
        """,
        "html": False
    })

    # Schedule follow-up research in 30 days
    await agent.call_tool("mcp_set_timer", {
        "duration_seconds": 30 * 24 * 3600,  # 30 days
        "timer_name": f"Follow-up: {competitor_url}",
        "callback_message": f"Time to re-check {competitor_url}"
    })

    print("✅ Research complete!")

---

Delayed Task Execution

Use timers for delayed or scheduled operations.

async def scheduled_report(agent, report_type, delay_hours=24):
    """Generate and send report after a delay."""

    # Schedule report generation
    timer = await agent.call_tool("mcp_set_timer", {
        "duration_seconds": delay_hours * 3600,
        "timer_name": f"{report_type} Report",
        "callback_message": f"Generate {report_type} report",
        "callback_data": {"report_type": report_type}
    })

    print(f"📅 Report scheduled for {delay_hours} hours from now")
    print(f"   Timer ID: {timer['timer_id']}")

    # Send confirmation
    await agent.call_tool("mcp_send_slack_message", {
        "message": f"📊 {report_type} report scheduled for "
                   f"{delay_hours} hours from now\n"
                   f"Timer: {timer['timer_id']}"
    })

    return timer


async def recurring_backup_notification(agent):
    """Send backup reminders every week."""

    await agent.call_tool("mcp_set_recurring_timer", {
        "interval_seconds": 7 * 24 * 3600,  # 1 week
        "timer_name": "Weekly Backup Reminder",
        "callback_message": "Time to verify backups!",
        "max_occurrences": None  # Run indefinitely
    })

    print("✅ Weekly backup reminder configured")

---

Error Recovery Workflow

Handle errors with admin escalation.

async def resilient_task(agent, task_description):
    """Execute task with automatic retry and admin escalation."""

    max_retries = 3
    retry_count = 0

    while retry_count < max_retries:
        try:
            # Attempt task
            result = await agent.call_tool("mcp_browser_execute_task", {
                "task": task_description,
                "max_steps": 30
            })

            if result["success"]:
                # Success! Notify and return
                await agent.call_tool("mcp_send_slack_message", {
                    "message": f"✅ Task completed: {task_description}"
                })
                return result

            retry_count += 1

            if retry_count < max_retries:
                # Wait before retry
                wait_seconds = 60 * retry_count
                print(f"⏳ Retry {retry_count}/{max_retries} in {wait_seconds}s...")

                await agent.call_tool("mcp_set_timer", {
                    "duration_seconds": wait_seconds,
                    "timer_name": f"Retry {retry_count}"
                })

                # Actual wait
                await asyncio.sleep(wait_seconds)

        except Exception as e:
            print(f"❌ Error: {e}")
            retry_count += 1

    # All retries failed - escalate to admin
    print("🚨 All retries failed, requesting admin assistance...")

    admin_help = await agent.call_tool("mcp_request_admin_approval", {
        "request_message": f"Task failed after {max_retries} retries. Manual intervention needed?",
        "context": {
            "task": task_description,
            "retries": retry_count,
            "last_error": str(result.get("error", "Unknown"))
        },
        "urgent": True,
        "timeout_seconds": 1800
    })

    if admin_help["approved"]:
        # Admin will handle manually
        await agent.call_tool("mcp_send_email", {
            "to_email": "admin@example.com",
            "subject": "Task Requires Manual Intervention",
            "body": f"Task: {task_description}\n"
                    f"Failed after {max_retries} retries\n"
                    f"Admin notes: {admin_help.get('admin_notes', 'None')}"
        })

    return None

---

Tips for Effective Usage

1. Combine Tools: Use multiple tools together for powerful workflows
2. Error Handling: Always check success field in results
3. Timeouts: Set appropriate timeouts for HITL requests
4. Notifications: Use multiple channels for critical alerts
5. Timers: Leverage timers for retries and scheduled tasks
6. Screenshots: Take screenshots for audit trail
7. Admin Context: Provide rich context in HITL requests

---

For more examples, see client_example.py and quickstart.py.