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 (Mastodon, Discord, Telegram, Matrix, etc.).

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 for a complete example
• Check the base PlatformAdapter for interface documentation
• 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! 🎉