DEV Community

johann
johann

Posted on

How to Connect AI Agents (Claude & Cursor) to Real-World Notifications with a Resilient MCP Server in Python

#mcp #automation #python #ai

How to Connect AI Agents (Claude & Cursor) to Real-World Notifications with a Resilient MCP Server in Python

Autonomous AI agents running in Claude Desktop, Cursor, Windsurf, or Antigravity are transforming software engineering and sysadmin workflows. However, when an agent completes a 45-minute build, encounters a critical deployment exception, or discovers an uncommitted bug, it has no native, secure way to ping you.

Giving an LLM direct shell execution with raw curl commands is notoriously brittle:

  • Secrets and tokens leak into prompt histories.
  • Runaway agent loops can fire 100 requests in 5 seconds, instantly getting your Telegram or WhatsApp bot banned.
  • Momentary network drops or HTTP 429 rate limits cause agent workflows to crash and abort.

The standardized solution is Model Context Protocol (MCP), Anthropic's open protocol for connecting AI models to external tools and services.

In this tutorial, we will walk through how OmniBridge MCP was built in pure Python to deliver resilient, multi-channel notifications (Telegram, Discord, WhatsApp, Slack) with offline queueing.

Architecture: Why Resilient Queuing Matters

Most naive MCP integrations make direct, synchronous HTTP requests. If the endpoint is down or throttles the connection, the agent throws an unrecoverable tool call error.

OmniBridge implements an enterprise resilience layer:

[ AI Agent: Claude / Cursor / Antigravity ]
                      │  JSON-RPC 2.0 (stdio)
                      ▼
            [ OmniBridge MCP Server ]
                      │
           [ Token Bucket Rate Limiter ]
                      │
             [ Channel Dispatcher ]
     ┌────────────────┼────────────────┐
     ▼                ▼                ▼
[ Telegram ]     [ Discord ]     [ WhatsApp Cloud ]
     │                │                │
     └────────────────┴────────────────┘
                      │
           (On API Timeout or 429)
                      ▼
     [ Persistent SQLite WAL Queue ]
                      │
       (Exponential Backoff Worker)
Enter fullscreen mode Exit fullscreen mode

1. Implementing the MCP Server Loop (stdio)

Rather than adding heavy framework dependencies, an MCP server can be built with standard JSON-RPC 2.0 over standard input and output (sys.stdin / sys.stdout).

import sys
import json
import logging

logging.basicConfig(level=logging.INFO, stream=sys.stderr)

class MCPServer:
    def __init__(self):
        self.server_name = "omnibridge-mcp"
        self.server_version = "1.0.0"

    def handle_request(self, req):
        method = req.get("method")
        req_id = req.get("id")

        if method == "initialize":
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {"tools": {}},
                    "serverInfo": {"name": self.server_name, "version": self.server_version}
                }
            }
        elif method == "tools/list":
            return {
                "jsonrpc": "2.0",
                "id": req_id,
                "result": {"tools": self.get_tools()}
            }
        # Dispatches tools/call...
Enter fullscreen mode Exit fullscreen mode

2. Real-Time Alert Dispatchers (Telegram & Discord)

Using httpx, we deliver rich messages to Telegram chats and Discord channels:

import httpx

def send_telegram_alert(bot_token: str, chat_id: str, text: str):
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {"chat_id": chat_id, "text": text, "parse_mode": "Markdown"}
    with httpx.Client(timeout=10.0) as client:
        return client.post(url, json=payload).json()
Enter fullscreen mode Exit fullscreen mode

3. Configuring Claude Desktop & Cursor

Connect the server in your claude_desktop_config.json:

{
  "mcpServers": {
    "omnibridge": {
      "command": "python",
      "args": ["-m", "omnibridge.server"],
      "cwd": "/path/to/omnibridge-mcp",
      "env": {
        "TELEGRAM_BOT_TOKEN": "your_bot_token",
        "DISCORD_WEBHOOK_URL": "https://discord.com/api/webhooks/..."
      }
    }
  }
}
Enter fullscreen mode Exit fullscreen mode

Now you can instruct Claude or Cursor:

"Run the build pipeline. If any test fails, notify me immediately on Telegram with the error trace."

4. Production Grade: WhatsApp Cloud API & Offline WAL Queue

For business-critical agent setups, missing notifications is not an option. The production tier features:

  • WhatsApp Cloud API Integration: Direct messaging with international number formatting.
  • SQLite Write-Ahead Logging (WAL): If an API is temporarily offline, the message is safely enqueued and retried with exponential backoff: delay = (2 ** retry_count) * 5 seconds
  • Token Bucket Rate Limiting: Throttling requests to prevent Meta/Discord account suspension.
  • Production Docker Compose: Ready to deploy as a sidecar container.

Resources and Downloads

  • Lite Version (Open Source): Available for free with Telegram and Discord support.
  • Enterprise Pro Edition ($14.99): Includes full WhatsApp Cloud API, Slack Block Kit, Matrix, custom HMAC webhooks, Docker Compose, and automated test suite.

👉 Get OmniBridge Pro on Gumroad ($14.99)

Top comments (0)