agentic-govbot/PLATFORMS.md

Platform Implementation Guide

This guide explains how to implement Govbot adapters for new social/communication platforms.

Overview

Govbot uses a platform-agnostic architecture that separates governance logic from platform-specific code. This allows the same constitutional reasoning and governance processes to work across different platforms.

Currently Implemented:

• ✅ Mastodon - Instance-wide governance with full admin/moderation API
• ✅ Slack - Channel-scoped governance with Socket Mode

Planned:

• 🚧 Discord - Server-wide governance (see example skeleton below)
• 🚧 Telegram - Group governance
• 🚧 Matrix - Room governance

The key abstraction is the PlatformAdapter interface, which defines how Govbot interacts with any platform.

Architecture

┌─────────────────────────────────────┐
│      Governance Logic               │
│  (Constitution, Agent, Primitives)  │  ← Platform-agnostic
└──────────────┬──────────────────────┘
               │
┌──────────────▼──────────────────────┐
│      PlatformAdapter Interface      │  ← This guide
└──────────────┬──────────────────────┘
               │
        ┌──────┴──────┐
        │             │
   ┌────▼────┐  ┌────▼────┐
   │Mastodon │  │Your     │
   │Adapter  │  │Adapter  │ ← What you'll build
   └─────────┘  └─────────┘

Quick Start

To implement a new platform:

1. Create src/govbot/platforms/yourplatform.py
2. Subclass PlatformAdapter
3. Implement all abstract methods
4. Define platform-specific skills
5. Add configuration support
6. Test with the mock bot
7. Document setup instructions

Step-by-Step Guide

Step 1: Create the Adapter File

Create a new file: src/govbot/platforms/yourplatform.py

"""
YourPlatform adapter for Govbot.

Brief description of the platform and what governance features this enables.
"""

from typing import Callable, Optional, Dict, Any, List
import logging

from .base import (
    PlatformAdapter,
    PlatformMessage,
    PlatformSkill,
    SkillParameter,
    MessageVisibility,
)

logger = logging.getLogger("govbot.platforms.yourplatform")


class YourPlatformAdapter(PlatformAdapter):
    """YourPlatform implementation of platform adapter."""

    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)
        # Initialize platform-specific client
        # Store necessary credentials
        pass

Step 2: Implement Connection Methods

    def connect(self) -> bool:
        """
        Establish connection to YourPlatform.

        Returns:
            True if connection successful

        Raises:
            Exception: If connection fails
        """
        try:
            # Initialize your platform client
            # Authenticate
            # Verify credentials
            # Store bot user ID and username

            self.connected = True
            self.bot_user_id = "your_bot_id"
            self.bot_username = "your_bot_name"

            logger.info(f"Connected to YourPlatform as {self.bot_username}")
            return True

        except Exception as e:
            logger.error(f"Connection failed: {e}")
            raise

    def disconnect(self):
        """Clean up and disconnect."""
        # Close connections
        # Stop listeners
        # Free resources
        self.connected = False
        logger.info("Disconnected from YourPlatform")

Step 3: Implement Message Listening

This is the core of how your bot receives messages:

    def start_listening(self, callback: Callable[[PlatformMessage], None]):
        """
        Start listening for messages mentioning the bot.

        Args:
            callback: Function to call with each message
        """
        if not self.connected:
            raise RuntimeError("Must connect() before listening")

        logger.info("Starting message listener")

        # Platform-specific listening implementation
        # Examples:
        # - WebSocket/streaming connection
        # - Polling API endpoint
        # - Webhook receiver

        def on_message(native_message):
            # Convert platform message to PlatformMessage
            normalized = self._normalize_message(native_message)

            # Filter for messages mentioning the bot
            if self._mentions_bot(normalized):
                callback(normalized)

        # Start your listener (may be async, threaded, etc.)
        # your_platform.listen(on_message)

    def _normalize_message(self, native_message) -> PlatformMessage:
        """
        Convert platform-native message format to PlatformMessage.

        This is crucial for platform abstraction!
        """
        return PlatformMessage(
            id=native_message["id"],
            text=self._extract_text(native_message),
            author_id=native_message["author"]["id"],
            author_handle=native_message["author"]["username"],
            timestamp=native_message["created_at"],
            thread_id=native_message.get("thread_id"),
            reply_to_id=native_message.get("reply_to"),
            visibility=self._map_visibility(native_message),
            mentions_bot=True,
            raw_data=native_message,  # Keep original for reference
        )

    def _mentions_bot(self, message: PlatformMessage) -> bool:
        """Check if message mentions the bot."""
        # Platform-specific mention detection
        # Examples:
        # - Check mentions list
        # - Search for @botname in text
        # - Check if in DM
        pass

Step 4: Implement Posting

    def post(
        self,
        message: str,
        thread_id: Optional[str] = None,
        reply_to_id: Optional[str] = None,
        visibility: MessageVisibility = MessageVisibility.PUBLIC,
    ) -> str:
        """
        Post a message to the platform.

        Args:
            message: Text content
            thread_id: Thread/channel to post in
            reply_to_id: Message to reply to
            visibility: Visibility level

        Returns:
            Message ID of posted message
        """
        if not self.connected:
            raise RuntimeError("Must connect() before posting")

        # Map abstract visibility to platform visibility
        platform_visibility = self._map_visibility_to_platform(visibility)

        # Use platform API to post
        result = your_platform.post_message(
            text=message,
            channel=thread_id,
            reply_to=reply_to_id,
            visibility=platform_visibility,
        )

        logger.info(f"Posted message {result['id']}")
        return result["id"]

Step 5: Define Platform Skills

Skills are platform-specific actions the bot can perform. Think about:

• Moderation: Ban, mute, delete content
• Administration: Change settings, update rules
• User management: Roles, permissions
• Content management: Pins, announcements

    def get_skills(self) -> List[PlatformSkill]:
        """Define YourPlatform-specific governance skills."""
        return [
            PlatformSkill(
                name="ban_user",
                description="Ban a user from the server/instance",
                category="moderation",
                parameters=[
                    SkillParameter(
                        "user_id",
                        "str",
                        "User ID to ban",
                        required=True
                    ),
                    SkillParameter(
                        "reason",
                        "str",
                        "Reason for ban",
                        required=True
                    ),
                    SkillParameter(
                        "duration",
                        "int",
                        "Ban duration in days (0 = permanent)",
                        required=False,
                        default=0
                    ),
                ],
                requires_confirmation=True,
                reversible=True,
                constitutional_authorization="Requires moderation authority",
            ),

            PlatformSkill(
                name="update_rules",
                description="Update server/instance rules",
                category="admin",
                parameters=[
                    SkillParameter(
                        "rules",
                        "list",
                        "List of rule texts",
                        required=True
                    ),
                ],
                requires_confirmation=True,
                reversible=True,
                constitutional_authorization="Requires constitutional process",
            ),

            # Add more skills specific to your platform
        ]

Step 6: Implement Skill Execution

    def execute_skill(
        self,
        skill_name: str,
        parameters: Dict[str, Any],
        actor: str,
    ) -> Dict[str, Any]:
        """Execute a platform skill."""

        if not self.connected:
            raise RuntimeError("Must connect() before executing skills")

        # Validate
        is_valid, error = self.validate_skill_execution(skill_name, parameters)
        if not is_valid:
            raise ValueError(error)

        logger.info(f"Executing {skill_name} for {actor}")

        # Route to skill handler
        if skill_name == "ban_user":
            return self._ban_user(parameters)
        elif skill_name == "update_rules":
            return self._update_rules(parameters)
        else:
            raise ValueError(f"Unknown skill: {skill_name}")

    def _ban_user(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Ban a user (example skill implementation)."""
        user_id = params["user_id"]
        reason = params["reason"]
        duration = params.get("duration", 0)

        # Use platform API to ban user
        your_platform.ban(user_id, reason=reason, duration=duration)

        return {
            "success": True,
            "message": f"Banned user {user_id}",
            "data": {"user_id": user_id, "reason": reason, "duration": duration},
            "reversible": True,
            "reverse_params": {"user_id": user_id},  # For unban
        }

Step 7: Implement User Info

    def get_user_info(self, user_id: str) -> Optional[Dict[str, Any]]:
        """Get information about a user."""
        if not self.connected:
            return None

        try:
            user = your_platform.get_user(user_id)

            return {
                "id": user["id"],
                "handle": user["username"],
                "display_name": user["display_name"],
                "roles": self._get_user_roles(user),
                "is_bot": user.get("bot", False),
            }
        except Exception as e:
            logger.error(f"Failed to get user info: {e}")
            return None

    def _get_user_roles(self, user) -> List[str]:
        """Determine user roles for governance."""
        roles = ["member"]

        if user.get("is_admin"):
            roles.append("admin")
        if user.get("is_moderator"):
            roles.append("moderator")

        return roles

Step 8: Implement Thread URLs

    def format_thread_url(self, thread_id: str) -> str:
        """Generate URL to view a thread."""
        return f"https://yourplatform.com/thread/{thread_id}"

Platform-Specific Considerations

Threading/Conversations

Different platforms handle conversations differently:

• Mastodon: Reply chains with in_reply_to_id
• Discord: Channels + threads
• Telegram: Groups + reply-to
• Slack: Channels + thread_ts

Map your platform's concept to thread_id in PlatformMessage.

Visibility/Privacy

Map your platform's privacy levels to MessageVisibility:

def _map_visibility_to_platform(self, visibility: MessageVisibility) -> str:
    """Map abstract visibility to platform-specific."""
    mapping = {
        MessageVisibility.PUBLIC: "your_platform_public",
        MessageVisibility.UNLISTED: "your_platform_unlisted",
        MessageVisibility.FOLLOWERS: "your_platform_followers",
        MessageVisibility.DIRECT: "your_platform_dm",
        MessageVisibility.PRIVATE: "your_platform_private",
    }
    return mapping.get(visibility, "your_platform_default")

Authentication

Different platforms have different auth:

• OAuth: Mastodon, Discord
• Bot tokens: Telegram, Discord, Slack
• API keys: Custom platforms

Store credentials securely in config:

def __init__(self, config: Dict[str, Any]):
    super().__init__(config)

    # Get required credentials
    self.api_token = config.get("api_token")
    self.bot_id = config.get("bot_id")

    if not self.api_token:
        raise ValueError("YourPlatform requires 'api_token' in config")

Rate Limiting

Handle platform rate limits gracefully:

import time
from functools import wraps

def rate_limited(max_per_minute):
    """Decorator for rate limiting."""
    min_interval = 60.0 / max_per_minute
    last_called = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = min_interval - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator

@rate_limited(max_per_minute=30)
def post(self, message, ...):
    # Your post implementation
    pass

Configuration

Add Config Model

In src/govbot/utils/config.py, add your platform config:

class YourPlatformConfig(BaseModel):
    """YourPlatform configuration"""

    api_token: str = Field(..., description="API token for bot")
    server_id: str = Field(..., description="Server/guild ID")
    # Add other required fields

class PlatformConfig(BaseModel):
    type: str = Field(...)
    mastodon: Optional[MastodonConfig] = None
    yourplatform: Optional[YourPlatformConfig] = None  # Add this

Update Example Config

In config/config.example.yaml:

platform:
  type: yourplatform

  yourplatform:
    api_token: your_token_here
    server_id: your_server_id
    # Other settings

Register in Bot

In src/govbot/bot.py, add your adapter:

from .platforms.yourplatform import YourPlatformAdapter

def _create_platform_adapter(self) -> PlatformAdapter:
    platform_type = self.config.platform.type.lower()

    if platform_type == "mastodon":
        return MastodonAdapter(self.config.platform.mastodon.model_dump())
    elif platform_type == "yourplatform":
        return YourPlatformAdapter(self.config.platform.yourplatform.model_dump())
    # ...

Testing Your Adapter

Unit Tests

Create tests/test_yourplatform.py:

import pytest
from src.govbot.platforms.yourplatform import YourPlatformAdapter
from src.govbot.platforms.base import PlatformMessage, MessageVisibility

def test_connection():
    """Test connecting to platform."""
    config = {
        "api_token": "test_token",
        "server_id": "test_server",
    }
    adapter = YourPlatformAdapter(config)

    # Mock the platform connection
    # Test that connect() returns True
    # Verify bot_user_id and bot_username are set

def test_message_normalization():
    """Test converting platform messages to PlatformMessage."""
    adapter = YourPlatformAdapter({...})

    native_message = {
        # Your platform's message format
    }

    normalized = adapter._normalize_message(native_message)

    assert isinstance(normalized, PlatformMessage)
    assert normalized.text == "expected text"
    assert normalized.author_handle == "expected_user"

def test_skills():
    """Test that skills are properly defined."""
    adapter = YourPlatformAdapter({...})
    skills = adapter.get_skills()

    assert len(skills) > 0
    assert all(skill.name for skill in skills)
    assert all(skill.category in ["admin", "moderation", "content", "user_management"]
               for skill in skills)

Integration Testing

1. Use Mock Mode: Test governance logic without platform
2. Sandbox Account: Create test account on your platform
3. Test Instance: Use development/test server if available

Manual Testing Checklist

• Bot can connect and authenticate
• Bot receives mentions
• Bot can post responses
• Bot can post in threads
• Visibility levels work correctly
• Skills execute successfully
• Rate limiting works
• Errors are handled gracefully
• Disconnection is clean

Common Patterns

Async vs Sync

If your platform client is async:

import asyncio
import threading

def start_listening(self, callback):
    """Start async listener in separate thread."""

    async def async_listen():
        async for message in your_platform.stream():
            normalized = self._normalize_message(message)
            callback(normalized)  # Callback is sync

    def run_async_loop():
        asyncio.run(async_listen())

    self.listener_thread = threading.Thread(target=run_async_loop, daemon=True)
    self.listener_thread.start()

Webhooks

If your platform uses webhooks instead of streaming:

from flask import Flask, request

class YourPlatformAdapter(PlatformAdapter):
    def __init__(self, config):
        super().__init__(config)
        self.app = Flask(__name__)
        self.callback = None

    def start_listening(self, callback):
        """Set up webhook receiver."""
        self.callback = callback

        @self.app.route("/webhook", methods=["POST"])
        def webhook():
            data = request.json
            message = self._normalize_message(data)
            if self.callback:
                self.callback(message)
            return {"status": "ok"}

        # Run Flask in separate thread
        threading.Thread(
            target=lambda: self.app.run(port=5000),
            daemon=True
        ).start()

Error Recovery

Implement reconnection logic:

def start_listening(self, callback):
    """Listen with automatic reconnection."""

    def listen_with_retry():
        while self.connected:
            try:
                your_platform.stream(on_message)
            except Exception as e:
                logger.error(f"Stream error: {e}")
                if self.connected:
                    logger.info("Reconnecting in 5 seconds...")
                    time.sleep(5)

    threading.Thread(target=listen_with_retry, daemon=True).start()

Documentation

User-Facing Documentation

Create docs/platforms/yourplatform.md:

# YourPlatform Setup Guide

## Prerequisites
- YourPlatform account
- Bot account created
- Admin access (for some features)

## Setup Steps

1. Create bot account at https://yourplatform.com/bots
2. Generate API token
3. Copy `config/config.example.yaml` to `config/config.yaml`
4. Update configuration:
   ```yaml
   platform:
     type: yourplatform
     yourplatform:
       api_token: YOUR_TOKEN
       server_id: YOUR_SERVER

5. Run the bot: python -m src.govbot

Available Features

• Governance proposals
• Voting in threads
• Moderation actions
• Admin commands

Permissions Required

• Read messages
• Post messages
• Manage server (for admin skills)

## Example: Discord Adapter Skeleton

Here's a skeleton for a Discord adapter:

```python
"""Discord platform adapter for Govbot."""

import discord
from discord.ext import commands
from typing import Callable, Optional, Dict, Any, List
import logging

from .base import (
    PlatformAdapter,
    PlatformMessage,
    PlatformSkill,
    SkillParameter,
    MessageVisibility,
)

logger = logging.getLogger("govbot.platforms.discord")


class DiscordAdapter(PlatformAdapter):
    """Discord implementation of platform adapter."""

    def __init__(self, config: Dict[str, Any]):
        super().__init__(config)

        token = config.get("token")
        self.guild_id = config.get("guild_id")

        if not token or not self.guild_id:
            raise ValueError("Discord requires 'token' and 'guild_id'")

        intents = discord.Intents.default()
        intents.message_content = True

        self.bot = commands.Bot(command_prefix="!", intents=intents)
        self.token = token
        self.message_callback = None

    def connect(self) -> bool:
        """Connect to Discord."""
        @self.bot.event
        async def on_ready():
            logger.info(f"Connected as {self.bot.user}")
            self.bot_user_id = str(self.bot.user.id)
            self.bot_username = self.bot.user.name
            self.connected = True

        # Run bot in thread
        import threading
        threading.Thread(
            target=lambda: self.bot.run(self.token),
            daemon=True
        ).start()

        # Wait for connection
        import time
        timeout = 10
        while not self.connected and timeout > 0:
            time.sleep(0.5)
            timeout -= 0.5

        return self.connected

    def disconnect(self):
        """Disconnect from Discord."""
        import asyncio
        asyncio.run(self.bot.close())
        self.connected = False

    def start_listening(self, callback: Callable[[PlatformMessage], None]):
        """Listen for mentions."""
        self.message_callback = callback

        @self.bot.event
        async def on_message(message):
            if message.author.id == self.bot.user.id:
                return

            if self.bot.user in message.mentions:
                normalized = self._normalize_message(message)
                if self.message_callback:
                    self.message_callback(normalized)

    def post(self, message: str, thread_id: Optional[str] = None,
             reply_to_id: Optional[str] = None,
             visibility: MessageVisibility = MessageVisibility.PUBLIC) -> str:
        """Post to Discord."""
        import asyncio

        async def send():
            channel = self.bot.get_channel(int(thread_id))
            msg = await channel.send(message)
            return str(msg.id)

        return asyncio.run(send())

    def get_skills(self) -> List[PlatformSkill]:
        """Discord-specific skills."""
        return [
            PlatformSkill(
                name="ban_user",
                description="Ban user from server",
                category="moderation",
                parameters=[
                    SkillParameter("user_id", "str", "User to ban"),
                    SkillParameter("reason", "str", "Ban reason"),
                ],
                requires_confirmation=True,
                reversible=True,
            ),
            # More Discord skills...
        ]

    def execute_skill(self, skill_name: str, parameters: Dict[str, Any],
                     actor: str) -> Dict[str, Any]:
        """Execute Discord skill."""
        # Implementation...
        pass

    def get_user_info(self, user_id: str) -> Optional[Dict[str, Any]]:
        """Get Discord user info."""
        # Implementation...
        pass

    def format_thread_url(self, thread_id: str) -> str:
        """Format Discord channel URL."""
        return f"https://discord.com/channels/{self.guild_id}/{thread_id}"

    def _normalize_message(self, msg: discord.Message) -> PlatformMessage:
        """Convert Discord message to PlatformMessage."""
        return PlatformMessage(
            id=str(msg.id),
            text=msg.content,
            author_id=str(msg.author.id),
            author_handle=msg.author.name,
            timestamp=msg.created_at,
            thread_id=str(msg.channel.id),
            reply_to_id=str(msg.reference.message_id) if msg.reference else None,
            visibility=MessageVisibility.PUBLIC,
            mentions_bot=True,
            raw_data=msg,
        )

Getting Help

• Review the Mastodon adapter (src/govbot/platforms/mastodon.py) for an instance-wide governance example
• Review the Slack adapter (src/govbot/platforms/slack.py) for a channel-scoped governance example
• Check the base PlatformAdapter (src/govbot/platforms/base.py) for interface documentation
• See setup guides: MASTODON_SETUP.md, SLACK_SETUP.md
• Look at test files for testing patterns
• Ask questions in project discussions

Contributing Your Adapter

Once your adapter works:

1. Add comprehensive docstrings
2. Write tests
3. Document setup in docs/platforms/
4. Update README.md with supported platforms
5. Submit a pull request

Thank you for extending Govbot to new platforms! 🎉