PLUGIN_DEVELOPMENT_GUIDE.md

SocialPie Plugin Development Guide

Welcome to SocialPie plugin development! This guide will help you create powerful plugins to extend SocialPie's functionality.

Table of Contents

1. Quick Start
2. Plugin Types
3. Plugin Structure
4. Hooks System
5. Commands
6. Events
7. Configuration
8. Testing
9. Publishing
10. Best Practices

---

Quick Start

1. Create a Plugin

# Install SocialPie with dev tools
pip install socialpie[dev]

# Create plugin from template
socialpie plugin create my-first-plugin --template server

# This creates:
# my-first-plugin/
# ├── __init__.py
# ├── plugin.py
# ├── plugin.yaml
# ├── README.md
# └── tests/
#     └── test_plugin.py

2. Basic Plugin Code

# plugin.py
from socialpie.plugins import ServerPlugin, hook, command
from socialpie.core.protocol import Message
from typing import override

class MyFirstPlugin(ServerPlugin):
    """My first SocialPie plugin!"""

    async def on_load(self) -> None:
        """Called when plugin loads"""
        self.logger.info("Plugin loaded!")
        self.message_count = 0

    @hook("after_message")
    async def count_messages(self, message: Message) -> None:
        """Count all messages"""
        self.message_count += 1
        self.logger.info(f"Total messages: {self.message_count}")

    @command("count", description="Show message count")
    async def show_count(self, client: Client, args: list[str]) -> str:
        """Handle /count command"""
        return f"Messages processed: {self.message_count}"

3. Plugin Manifest

# plugin.yaml
manifest:
  name: my-first-plugin
  version: 0.1.0
  author: Your Name
  description: My first SocialPie plugin
  license: MIT
  type: server
  entry_point: my_first_plugin.plugin:MyFirstPlugin

  requires: ">=1.0.0,<2.0.0"

  permissions:
    - messages.read

config:
  enabled: true

4. Test Your Plugin

# Run in development mode with hot-reload
socialpie server --dev --plugin ./my-first-plugin

# Or run tests
cd my-first-plugin
pytest

---

Plugin Types

SocialPie supports multiple plugin types:

Server Plugin

Runs on the server, processes messages, manages connections.

from socialpie.plugins import ServerPlugin

class MyServerPlugin(ServerPlugin):
    @hook("after_message")
    async def process(self, message: Message) -> None:
        pass

Client Plugin

Runs on the client, enhances UI, processes local events.

from socialpie.plugins import ClientPlugin

class MyClientPlugin(ClientPlugin):
    @hook("before_send")
    async def modify_outgoing(self, text: str) -> str:
        return text.upper()

UI Plugin

Adds custom UI components, themes, layouts.

from socialpie.plugins import UIPlugin

class MyThemePlugin(UIPlugin):
    @hook("register_themes")
    def register_themes(self, registry):
        registry.register(my_theme)

Hybrid Plugin

Works on both server and client.

from socialpie.plugins import Plugin

class MyHybridPlugin(Plugin):
    @hook("after_message")  # Works on both
    async def process(self, message: Message) -> None:
        pass

---

Plugin Structure

Minimal Plugin

my-plugin/
├── __init__.py
├── plugin.py         # Main plugin code
└── plugin.yaml       # Manifest

Full Plugin

my-plugin/
├── __init__.py
├── plugin.py         # Main plugin code
├── plugin.yaml       # Manifest
├── config.py         # Configuration models
├── handlers.py       # Message handlers
├── commands.py       # Command handlers
├── utils.py          # Utility functions
├── README.md         # Documentation
├── LICENSE           # License file
├── requirements.txt  # Dependencies
├── tests/            # Tests
│   ├── __init__.py
│   ├── test_plugin.py
│   └── test_commands.py
└── assets/           # Static files
    ├── icon.png
    └── config_schema.json

---

Hooks System

Hooks let you intercept and modify behavior at specific points.

Available Server Hooks

# Lifecycle hooks
@hook("server_start")
async def on_start(self) -> None:
    """Server starting"""
    pass

@hook("server_stop")
async def on_stop(self) -> None:
    """Server stopping"""
    pass

# Connection hooks
@hook("connection_open")
async def on_connect(self, client: Client) -> None:
    """Client connected"""
    pass

@hook("connection_close")
async def on_disconnect(self, client: Client) -> None:
    """Client disconnected"""
    pass

# Message hooks
@hook("before_message")
async def before_msg(self, message: Message) -> Message | None:
    """Intercept BEFORE processing - can modify or block"""
    message.content = message.content.upper()
    return message  # Return modified or None to block

@hook("after_message")
async def after_msg(self, message: Message) -> None:
    """Process AFTER message sent"""
    self.log_message(message)

@hook("broadcast_message")
async def on_broadcast(self, message: Message) -> None:
    """Message being broadcast to all clients"""
    pass

Hook Priority

Control execution order with priority (higher runs first):

@hook("before_message", priority=100)
async def high_priority(self, message: Message) -> Message:
    """Runs first"""
    return message

@hook("before_message", priority=50)
async def medium_priority(self, message: Message) -> Message:
    """Runs second"""
    return message

@hook("before_message", priority=1)
async def low_priority(self, message: Message) -> Message:
    """Runs last"""
    return message

Available Client Hooks

@hook("before_send")
async def before_send(self, text: str) -> str | None:
    """Modify message before sending"""
    return text

@hook("after_receive")
async def after_receive(self, message: Message) -> None:
    """Process received message"""
    pass

@hook("ui_render")
async def on_render(self, ui_state: dict) -> None:
    """UI rendering"""
    pass

@hook("connected")
async def on_connect(self) -> None:
    """Connected to server"""
    pass

@hook("disconnected")
async def on_disconnect(self) -> None:
    """Disconnected from server"""
    pass

---

Commands

Add custom chat commands that users can invoke.

Basic Command

@command("hello", description="Say hello")
async def hello_command(
    self,
    client: Client,
    args: list[str]
) -> str:
    """Handle /hello command"""
    return "Hello, world!"

Command with Arguments

@command("greet", description="Greet someone")
async def greet_command(
    self,
    client: Client,
    args: list[str]
) -> str:
    """Handle /greet <name>"""
    if not args:
        return "Usage: /greet <name>"

    name = " ".join(args)
    return f"Hello, {name}! 👋"

Advanced Command

from socialpie.plugins import command, CommandContext

@command(
    "kick",
    description="Kick a user",
    usage="/kick <username> [reason]",
    permissions=["users.kick"],
    aliases=["k"]
)
async def kick_command(
    self,
    ctx: CommandContext
) -> str:
    """Kick a user from the server"""
    if not ctx.args:
        return "Usage: /kick <username> [reason]"

    username = ctx.args[0]
    reason = " ".join(ctx.args[1:]) if len(ctx.args) > 1 else "No reason"

    # Check permission
    if not ctx.client.has_permission("users.kick"):
        return "You don't have permission to kick users"

    # Kick the user
    user = self.server.get_user(username)
    if user:
        await self.server.disconnect_user(user, reason)
        return f"Kicked {username}: {reason}"
    else:
        return f"User '{username}' not found"

---

Events

The event system allows pub/sub style communication between plugins.

Listen to Events

from socialpie.plugins import event_listener
from socialpie.core.events import Event

@event_listener("user_joined")
async def on_user_joined(self, event: Event) -> None:
    """Called when a user joins"""
    username = event.data["username"]
    await self.send_welcome_message(username)

@event_listener("custom_event")
async def on_custom(self, event: Event) -> None:
    """Listen to custom events from other plugins"""
    self.logger.info(f"Custom event: {event.data}")

Emit Events

from socialpie.core.events import Event
from datetime import datetime

# Emit event for other plugins
await self.emit_event(Event(
    type="custom_event",
    timestamp=datetime.now(),
    source=self.name,
    data={
        "key": "value",
        "number": 42
    }
))

Built-in Events

# User events
"user_joined"     # User connected
"user_left"       # User disconnected
"user_banned"     # User was banned

# Message events
"message_sent"    # Message sent
"message_edited"  # Message edited
"message_deleted" # Message deleted

# System events
"server_started"  # Server started
"server_stopped"  # Server stopped
"plugin_loaded"   # Plugin loaded
"plugin_unloaded" # Plugin unloaded

---

Configuration

Define Config Schema

# config.py
from pydantic import BaseModel, Field

class PluginConfig(BaseModel):
    """Plugin configuration"""
    enabled: bool = True
    log_file: str = Field(default="~/.socialpie/logs/plugin.log")
    max_size: int = Field(default=100, ge=1, le=1000)
    categories: list[str] = Field(default_factory=list)

Use Configuration

from .config import PluginConfig

class MyPlugin(ServerPlugin):
    def __init__(self, config: dict) -> None:
        super().__init__(config)
        # Validate and parse config
        self.cfg = PluginConfig(**config)

    async def on_load(self) -> None:
        self.logger.info(f"Log file: {self.cfg.log_file}")
        self.logger.info(f"Max size: {self.cfg.max_size}")

plugin.yaml Config

config:
  enabled: true
  log_file: /var/log/socialpie/plugin.log
  max_size: 200
  categories:
    - general
    - admin
    - fun

config_schema:
  type: object
  properties:
    enabled:
      type: boolean
      description: Enable the plugin
    log_file:
      type: string
      description: Path to log file
    max_size:
      type: integer
      minimum: 1
      maximum: 1000
      description: Maximum size
    categories:
      type: array
      items:
        type: string
      description: Categories to process

---

Testing

Test Structure

# tests/test_plugin.py
import pytest
from socialpie.plugins.testing import PluginTestCase
from my_plugin.plugin import MyPlugin

class TestMyPlugin(PluginTestCase):
    plugin_class = MyPlugin

    async def test_message_hook(self):
        """Test message processing"""
        message = self.create_message(
            content="test",
            user="testuser"
        )

        result = await self.plugin.process_message(message)

        assert result.content == "TEST"

    async def test_command(self):
        """Test command execution"""
        client = self.create_client("testuser")
        response = await self.execute_command("/mycommand arg1", client)

        assert "success" in response.lower()

    async def test_event(self):
        """Test event handling"""
        event = self.create_event("custom_event", {"key": "value"})

        await self.plugin.handle_event(event)

        assert self.plugin.event_received

Run Tests

# Run all tests
pytest

# Run with coverage
pytest --cov=my_plugin --cov-report=html

# Run specific test
pytest tests/test_plugin.py::TestMyPlugin::test_message_hook

# Run in watch mode
ptw

---

Publishing

1. Prepare Plugin

# Validate plugin
socialpie plugin validate ./my-plugin

# Run tests
cd my-plugin && pytest

# Lint code
ruff check .
mypy .

2. Build Package

# Build plugin package
socialpie plugin build ./my-plugin

# Creates: my-plugin-0.1.0.zip

3. Publish

# Publish to marketplace
socialpie plugin publish my-plugin

# Or create shareable package
socialpie plugin package my-plugin --output my-plugin.zip

4. Users Install

# From marketplace
socialpie plugin install my-plugin

# From file
socialpie plugin install my-plugin.zip

# From git
socialpie plugin install git+https://github.com/user/my-plugin

---

Best Practices

1. Type Hints

Always use type hints for better IDE support and error checking:

from typing import Optional
from socialpie.core.protocol import Message

@hook("before_message")
async def process(self, message: Message) -> Optional[Message]:
    return message

2. Error Handling

Handle errors gracefully:

@hook("after_message")
async def process(self, message: Message) -> None:
    try:
        result = await self.risky_operation(message)
    except ValueError as e:
        self.logger.error(f"Invalid value: {e}")
    except Exception as e:
        self.logger.exception(f"Unexpected error: {e}")

3. Async/Await

Use async for I/O operations:

@hook("after_message")
async def process(self, message: Message) -> None:
    # Good: async I/O
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()

    # Avoid: blocking I/O
    # response = requests.get(url)  # DON'T DO THIS

4. Resource Cleanup

Clean up resources properly:

async def on_load(self) -> None:
    self.db_connection = await create_connection()

async def on_unload(self) -> None:
    if self.db_connection:
        await self.db_connection.close()

5. Configuration Validation

Validate configuration early:

def __init__(self, config: dict) -> None:
    super().__init__(config)

    # Validate required fields
    if "api_key" not in config:
        raise ValueError("api_key is required")

    # Use Pydantic for complex validation
    self.cfg = MyConfig(**config)

6. Logging

Use proper logging levels:

self.logger.debug("Detailed debug info")
self.logger.info("General information")
self.logger.warning("Warning message")
self.logger.error("Error occurred")
self.logger.exception("Error with traceback")

7. Documentation

Document your plugin well:

class MyPlugin(ServerPlugin):
    """
    A plugin that does awesome things.

    This plugin processes messages and adds emoji reactions
    based on sentiment analysis.

    Configuration:
        enabled (bool): Whether plugin is enabled
        api_key (str): API key for sentiment service

    Commands:
        /emoji <message> - Suggest emoji for message

    Example:
        /emoji "I love this!"
        # Returns: 😍 ❤️ 🎉
    """

    @command("emoji", description="Suggest emoji for message")
    async def emoji_command(
        self,
        client: Client,
        args: list[str]
    ) -> str:
        """
        Suggest emoji based on message sentiment.

        Args:
            client: The client who sent command
            args: Command arguments

        Returns:
            Suggested emoji string
        """
        pass

8. Testing

Write comprehensive tests:

class TestMyPlugin(PluginTestCase):
    async def test_normal_case(self):
        """Test normal operation"""
        pass

    async def test_edge_cases(self):
        """Test edge cases"""
        pass

    async def test_error_handling(self):
        """Test error handling"""
        pass

    async def test_config_validation(self):
        """Test configuration validation"""
        pass

9. Performance

Optimize performance:

# Cache expensive operations
from functools import lru_cache

@lru_cache(maxsize=100)
def expensive_operation(self, key: str) -> str:
    return result

# Use async for concurrency
async def process_multiple(self, messages: list[Message]) -> None:
    tasks = [self.process_one(msg) for msg in messages]
    await asyncio.gather(*tasks)

10. Security

Follow security best practices:

# Validate input
@command("admin")
async def admin_command(self, client: Client, args: list[str]) -> str:
    # Check permissions
    if not client.has_permission("admin"):
        return "Permission denied"

    # Sanitize input
    safe_input = escape_html(args[0])

    # Use parameterized queries
    await db.execute("SELECT * FROM users WHERE id = ?", (user_id,))

---

Plugin Examples

Example 1: Welcome Bot

from socialpie.plugins import ServerPlugin, event_listener, hook
from socialpie.core.events import Event
from socialpie.core.protocol import Message
from datetime import datetime

class WelcomeBotPlugin(ServerPlugin):
    """Greets new users when they join"""

    @event_listener("user_joined")
    async def greet_user(self, event: Event) -> None:
        username = event.data["username"]

        welcome_msg = Message(
            type="system",
            user="WelcomeBot",
            content=f"Welcome to the server, {username}! 👋",
            timestamp=datetime.now(),
            metadata={}
        )

        await self.server.broadcast(welcome_msg)

Example 2: Chat Logger

from socialpie.plugins import ServerPlugin, hook
from socialpie.core.protocol import Message
import json
from pathlib import Path
from datetime import datetime

class ChatLoggerPlugin(ServerPlugin):
    """Logs all chat messages to file"""

    async def on_load(self) -> None:
        self.log_file = Path(self.config.get("log_file", "chat.log"))
        self.log_file.parent.mkdir(parents=True, exist_ok=True)

    @hook("after_message")
    async def log_message(self, message: Message) -> None:
        if message.type == "chat":
            log_entry = {
                "timestamp": message.timestamp.isoformat(),
                "user": message.user,
                "content": message.content
            }

            async with aiofiles.open(self.log_file, "a") as f:
                await f.write(json.dumps(log_entry) + "\n")

Example 3: Dice Roller

from socialpie.plugins import ServerPlugin, command
import random
import re

class DiceRollerPlugin(ServerPlugin):
    """Roll dice using /roll command"""

    @command("roll", description="Roll dice (e.g., /roll 2d6)")
    async def roll_dice(
        self,
        client: Client,
        args: list[str]
    ) -> str:
        if not args:
            return "Usage: /roll <dice> (e.g., 2d6, 1d20)"

        # Parse dice notation (e.g., "2d6")
        match = re.match(r"(\d+)d(\d+)([+\-]\d+)?", args[0])
        if not match:
            return "Invalid format. Use: XdY (e.g., 2d6, 1d20+5)"

        num_dice = int(match.group(1))
        sides = int(match.group(2))
        modifier = int(match.group(3) or 0)

        # Validate
        if num_dice > 100 or sides > 1000:
            return "That's too many dice or sides!"

        # Roll dice
        rolls = [random.randint(1, sides) for _ in range(num_dice)]
        total = sum(rolls) + modifier

        # Format output
        rolls_str = ", ".join(map(str, rolls))
        modifier_str = f" {modifier:+d}" if modifier != 0 else ""

        return f"🎲 [{rolls_str}]{modifier_str} = **{total}**"

---

Resources

• Full API Documentation: https://socialpie.dev/docs/api
• Plugin Examples: https://github.com/socialpie/plugins
• Community Plugins: https://socialpie.dev/marketplace
• Discord: https://discord.gg/socialpie
• GitHub: https://github.com/socialpie/socialpie

---

Getting Help

• Read the docs: https://socialpie.dev/docs
• Join Discord: https://discord.gg/socialpie
• Ask on GitHub Discussions: https://github.com/socialpie/socialpie/discussions
• File issues: https://github.com/socialpie/socialpie/issues

Happy plugin development! 🎉