Revia MCP Client

Python client for the Revia MCP bridge. Connect your Python code to WhatsApp, Telegram, Slack, and Gmail through a single async client. Also supports bidirectional chat with coding agents connected to your Revia instance.

Contents

• Install
• Quickstart
• Real-time events via WebSocket
• Chat with connected coding agents
• WebSocket client (agent-side)
• API Reference
• Token
• License

Install

pip install revia-mcp

Quickstart

import asyncio
from revia_mcp import ReviaMCPClient

async def main():
    async with ReviaMCPClient(
        "https://revia.devshub.ai/api/v1/mcp",
        token="rvagent_YOUR_TOKEN_HERE",
    ) as revia:

        # Health check
        pong = await revia.ping()
        print(pong)  # {"status": "pong", "user_id": "...", ...}

        # List channels
        channels = await revia.channels_list()
        for ch in channels["channels"]:
            print(f"{ch['platform']}: {'connected' if ch['reachable'] else 'offline'}")

        # Send a WhatsApp message
        await revia.messages_send(
            "whatsapp:974XXXXXXXX@s.whatsapp.net",
            "Hello from Python!",
        )

        # Read recent messages
        msgs = await revia.messages_read("whatsapp:974XXXXXXXX@s.whatsapp.net", limit=10)
        for m in msgs["messages"]:
            print(f"[{m['timestamp']}] {m.get('sender')}: {m.get('content')}")

        # Send an email
        await revia.email_send(
            to="client@example.com",
            subject="Meeting follow-up",
            body="Thanks for your time today!",
        )

asyncio.run(main())

Real-time events via WebSocket

async def stream_events():
    async with ReviaMCPClient(
        "https://revia.devshub.ai/api/v1/mcp",
        token="rvagent_YOUR_TOKEN_HERE",
    ) as revia:

        async for event in revia.events_ws():
            print(f"[{event['platform']}] {event['from']}: {event['content']}")

The WebSocket reconnects automatically on disconnect with exponential backoff. Pass name="MyBot" to register with a human-readable name so Revia can discover and chat with your agent.

Chat with connected coding agents

Revia can discover and chat with coding agents that are connected via WebSocket. Use agents_list to see who's online, then agent_chat to send a natural language message and get a response:

async def chat_with_agents():
    async with ReviaMCPClient(
        "https://revia.devshub.ai/api/v1/mcp",
        token="rvagent_YOUR_TOKEN_HERE",
    ) as revia:

        # See who's connected
        agents = await revia.agents_list()
        for a in agents["agents"]:
            print(f"  {a['name']} ({a['agent_id']})")

        # Send a natural language message and get a response
        reply = await revia.agent_chat(
            agent_id="agent_abc123",
            message="What's the status of the deployment?",
        )
        print(reply["response"])

The call blocks for up to 60 seconds waiting for the agent's response. Pass timeout_s to adjust.

WebSocket client (agent-side)

If you're building a coding agent that Revia should be able to chat with, use ReviaWebSocket directly. It handles auth, registration, event streaming, and chat responses:

import asyncio
from revia_mcp.ws import ReviaWebSocket

async def handle_chat(chat_id: str, message: str) -> str:
    """Revia sent us a message — respond in natural language."""
    if "deployment" in message.lower():
        return "Deployment is green — all pods healthy, last deploy 5 minutes ago."
    if "errors" in message.lower():
        return "No errors in the last hour. Error rate is 0.02%."
    return f"I received: {message}"

async def main():
    ws = ReviaWebSocket(
        "wss://revia.devshub.ai/api/v1/mcp/ws",
        token="rvagent_YOUR_TOKEN_HERE",
        name="My CodeSync Agent",
        on_chat=handle_chat,
    )

    async for event in ws:
        print(f"[{event.get('platform', 'system')}] {event.get('type')}: {event.get('content', '')}")

asyncio.run(main())

If you prefer manual control over chat responses, omit on_chat and handle chat events yourself:

async for event in ws:
    if event.get("type") == "chat":
        chat_id = event["chat_id"]
        message = event["message"]
        # ... think about it ...
        await ws.send_chat_response(chat_id, "Here's my response.")
    else:
        print(f"Event: {event}")

ReviaWebSocket constructor:

Properties and methods:

API Reference

Health

Channels & Contacts

Messages

Email

Revia AI

Events

Agents

Token

Generate an agent token in the Revia dashboard: Settings → Coding Agent (MCP) → Generate Token. Tokens use the rvagent_ prefix and are shown only once.

License

MIT