🤖 Discord.py × OMP Plugin Cheat Sheet #

// ~/.omp/agent/extensions/my-ext/index.ts
import type { ExtensionAPI } from "@oh-my-pi/pi-coding-agent";

export default function myExt(pi: ExtensionAPI) {
  // ── Registration Phase (sync) ──
  pi.setLabel("My Extension");

  pi.on("agent_end", async (event, ctx) => { /* ... */ });
  pi.on("session_start", async (event, ctx) => { /* ... */ });

  pi.registerTool({ name: "my_tool", /* ... */ });
  pi.registerCommand("my-cmd", { /* ... */ });

  // ⚠️ CANNOT call pi.sendMessage() during load — only in handlers
}

session_start → input → agent_start → turn_start
  → tool_call → tool_execution_start → tool_execution_end → tool_result
  → turn_end → agent_end → session_shutdown

import type { ExtensionAPI } from "@oh-my-pi/pi-coding-agent";

export default function discordNotify(pi: ExtensionAPI) {
  const WEBHOOK = process.env.DISCORD_WEBHOOK_URL;
  if (!WEBHOOK) return;

  const send = (content: string) =>
    fetch(WEBHOOK, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ content }),
    }).catch(() => {});

  // 🔔 Auto-notify when agent finishes
  pi.on("agent_end", async () => {
    await send(`✅ **Agent finished** — ${new Date().toLocaleTimeString()}`);
  });

  // 💬 Slash command: /discord-ping <message>
  pi.registerCommand("discord-ping", {
    description: "Send a message to the Discord webhook",
    handler: async (args, ctx) => {
      await send(args.trim() || "Ping from OMP!");
      ctx.ui.notify("Sent to Discord", "info");
    },
  });
}

# 1. Create webhook: Discord → Server Settings → Integrations → Webhooks → New
# 2. Export URL
export DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/123/abc..."

# 3. Drop file, restart OMP. Done.

OMP Agent ◀──stdio JSON-RPC──▶ Python MCP Server ◀──gateway──▶ Discord

#!/usr/bin/env python3
"""Discord bot exposed as MCP tool server."""
import asyncio, os
import discord
from discord.ext import commands
from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

# ── Bot ──
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)
ready = asyncio.Event()

@bot.event
async def on_ready():
    ready.set()

# ── MCP Server ──
app = Server("discord-mcp")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="discord_send",
            description="Send a message to a Discord channel",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel_id": {"type": "string"},
                    "content": {"type": "string"},
                },
                "required": ["channel_id", "content"],
            },
        ),
        types.Tool(
            name="discord_read",
            description="Read recent messages from a Discord channel",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel_id": {"type": "string"},
                    "limit": {"type": "integer", "default": 10},
                },
                "required": ["channel_id"],
            },
        ),
        types.Tool(
            name="discord_guilds",
            description="List all guilds the bot is in",
            inputSchema={"type": "object", "properties": {}},
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    await ready.wait()

    if name == "discord_send":
        ch = bot.get_channel(int(arguments["channel_id"]))
        if not ch:
            return [types.TextContent(type="text", text="Channel not found")]
        await ch.send(arguments["content"])
        return [types.TextContent(type="text", text=f"Sent to #{ch.name}")]

    if name == "discord_read":
        ch = bot.get_channel(int(arguments["channel_id"]))
        if not ch:
            return [types.TextContent(type="text", text="Channel not found")]
        msgs = []
        async for m in ch.history(limit=arguments.get("limit", 10)):
            msgs.append(f"[{m.created_at:%H:%M}] {m.author.display_name}: {m.content}")
        return [types.TextContent(type="text", text="\n".join(reversed(msgs)) or "(empty)")]

    if name == "discord_guilds":
        lines = [f"{g.name} (id={g.id})" for g in bot.guilds]
        return [types.TextContent(type="text", text="\n".join(lines) or "(none)")]

    return [types.TextContent(type="text", text=f"Unknown tool: {name}")]

async def main():
    token = os.environ["DISCORD_TOKEN"]
    async with bot:
        bot_task = asyncio.create_task(bot.start(token))
        async with stdio_server() as (r, w):
            await app.run(r, w, app.create_initialization_options())
        bot_task.cancel()

if __name__ == "__main__":
    asyncio.run(main())

{
  "mcpServers": {
    "discord": {
      "command": "python3",
      "args": ["/home/dev/.omp/tools/discord-mcp/server.py"],
      "env": { "DISCORD_TOKEN": "DISCORD_TOKEN" }
    }
  }
}

pip install discord.py mcp
# env value "DISCORD_TOKEN": "DISCORD_TOKEN" = copy from shell env
export DISCORD_TOKEN="your-bot-token"
# Restart OMP — it auto-discovers mcp.json and launches the server

┌──────────────────────────────────┐
│ OMP Extension (Bun/TS)          │
│ ├ on(agent_end) → POST /notify  │
│ ├ registerTool(discord_notify)  │
│ └ registerCommand(/discord)     │
└──────────┬───────────────────────┘
           │ HTTP :8901
┌──────────▼───────────────────────┐
│ discord.py Bot + aiohttp (Py)   │
│ ├ POST /notify → channel.send() │
│ ├ GET  /status → bot info       │
│ ├ GET  /channels → list all     │
│ └ !ask-omp <prompt> → omp CLI   │
└──────────┬───────────────────────┘
           │ Discord Gateway
┌──────────▼───────────────────────┐
│ Discord                         │
└──────────────────────────────────┘

#!/usr/bin/env python3
"""discord.py sidecar bot with HTTP API for OMP."""
import asyncio, logging, os, subprocess
import discord
from discord import app_commands
from discord.ext import commands
from aiohttp import web

log = logging.getLogger("discord-omp")
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix="!", intents=intents)

NOTIFY_CH = int(os.environ.get("DISCORD_NOTIFY_CHANNEL", "0"))
HTTP_PORT = int(os.environ.get("DISCORD_HTTP_PORT", "8901"))

# ── HTTP endpoints (called by OMP extension) ──
async def handle_notify(req: web.Request) -> web.Response:
    data = await req.json()
    ch = bot.get_channel(NOTIFY_CH)
    if not ch:
        return web.json_response({"error": "channel not found"}, status=404)

    embed_data = data.get("embed")
    if embed_data:
        embed = discord.Embed(
            title=embed_data.get("title", ""),
            description=embed_data.get("description", ""),
            color=embed_data.get("color", 0x5865F2),
        )
        for f in embed_data.get("fields", []):
            embed.add_field(name=f["name"], value=f["value"], inline=f.get("inline", False))
        await ch.send(content=data.get("content"), embed=embed)
    else:
        await ch.send(data.get("content", "📨 (empty)"))
    return web.json_response({"ok": True})

async def handle_status(_: web.Request) -> web.Response:
    return web.json_response({
        "bot": str(bot.user), "guilds": len(bot.guilds),
        "latency_ms": round(bot.latency * 1000, 1), "ready": bot.is_ready(),
    })

async def handle_channels(_: web.Request) -> web.Response:
    return web.json_response([
        {"id": str(c.id), "name": c.name, "guild": g.name}
        for g in bot.guilds for c in g.text_channels
    ])

async def start_http():
    app = web.Application()
    app.router.add_post("/notify", handle_notify)
    app.router.add_get("/status", handle_status)
    app.router.add_get("/channels", handle_channels)
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner, "127.0.0.1", HTTP_PORT).start()
    log.info(f"HTTP API → 127.0.0.1:{HTTP_PORT}")

# ── Discord commands ──
@bot.event
async def on_ready():
    await start_http()
    await bot.tree.sync()
    log.info(f"Ready as {bot.user}")

@bot.hybrid_command(name="ask-omp", description="Ask the OMP coding agent")
@app_commands.describe(prompt="Your question")
async def ask_omp(ctx: commands.Context, *, prompt: str):
    await ctx.defer()
    try:
        r = subprocess.run(
            ["omp", "-p", prompt, "--no-input"],
            capture_output=True, text=True, timeout=120, cwd="/home/dev",
        )
        out = r.stdout.strip()[:1900] or "(no output)"
        await ctx.send(f"```\n{out}\n```")
    except subprocess.TimeoutExpired:
        await ctx.send("⏱️ Timed out (120s)")
    except Exception as e:
        await ctx.send(f"❌ {e}")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(bot.start(os.environ["DISCORD_TOKEN"]))

import type { ExtensionAPI } from "@oh-my-pi/pi-coding-agent";

const BOT = process.env.DISCORD_BOT_URL ?? "http://127.0.0.1:8901";

export default function discordBridge(pi: ExtensionAPI) {
  const { z } = pi.zod;
  pi.setLabel("Discord Bridge");

  // 🔔 Auto-notify on agent completion
  pi.on("agent_end", async (_ev, ctx) => {
    const tokens = ctx.getContextUsage?.()?.tokens ?? "?";
    await fetch(`${BOT}/notify`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        embed: {
          title: "✅ Agent Run Complete",
          description: `Session: ${pi.getSessionName?.() ?? "—"}`,
          color: 0x00ff00,
          fields: [{ name: "Tokens", value: `${tokens}`, inline: true }],
        },
      }),
    }).catch(() => {});
  });

  // 🛠️ LLM-callable tool
  pi.registerTool({
    name: "discord_notify",
    label: "Discord Notify",
    description: "Send a notification to the configured Discord channel",
    parameters: z.object({
      message: z.string().describe("Message to send"),
    }),
    async execute(_id, params) {
      const res = await fetch(`${BOT}/notify`, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ content: params.message }),
      });
      const data = (await res.json()) as { ok?: boolean; error?: string };
      return data.ok
        ? { content: [{ type: "text", text: "✅ Sent to Discord" }] }
        : { content: [{ type: "text", text: `❌ ${data.error}` }], isError: true };
    },
  });

  // ⌨️ OMP slash command: /discord
  pi.registerCommand("discord", {
    description: "Check Discord bot status",
    handler: async (_args, ctx) => {
      try {
        const s = (await (await fetch(`${BOT}/status`)).json()) as Record<string, unknown>;
        ctx.ui.notify(`🤖 ${s.bot} | ${s.guilds} guilds | ${s.latency_ms}ms`, "info");
      } catch {
        ctx.ui.notify("Bot unreachable", "warning");
      }
    },
  });
}

# ~/.config/systemd/user/discord-omp-bot.service
[Unit]
Description=Discord OMP Bridge Bot
After=network.target

[Service]
ExecStart=/usr/bin/python3 /home/dev/discord_omp_bot.py
Environment=DISCORD_TOKEN=your-token
Environment=DISCORD_NOTIFY_CHANNEL=123456789
Environment=DISCORD_HTTP_PORT=8901
Restart=on-failure
RestartSec=5

[Install]
WantedBy=default.target

systemctl --user daemon-reload
systemctl --user enable --now discord-omp-bot
journalctl --user -u discord-omp-bot -f

omp-discord-bridge/
├── index.ts          ← source
├── package.json
└── dist/
    └── index.js      ← bundled (generated)

{
  "name": "omp-discord-bridge",
  "version": "1.0.0",
  "type": "module",
  "main": "dist/index.js",
  "omp": { "extensions": ["./dist/index.js"] },
  "peerDependencies": {
    "@oh-my-pi/pi-coding-agent": "*"
  },
  "scripts": {
    "build": "bun build ./index.ts --outdir ./dist --target node --external @oh-my-pi/pi-coding-agent"
  }
}

# Bundle (CRITICAL — OMP legacy loader copies to /tmp, breaking relative imports)
bun build ./index.ts --outdir ./dist --target node \
  --external @oh-my-pi/pi-coding-agent

# Install as plugin
omp plugin link /path/to/omp-discord-bridge

# Verify
omp plugin list

1. discord.com/developers/applications → New Application
2. Bot tab → Reset Token → 📋 copy
3. Bot tab → ✅ MESSAGE CONTENT Intent
4. OAuth2 → URL Generator → bot + applications.commands
   → Send Messages, Read History, Embed Links, Use Slash Commands
   → Copy invite URL → open in browser → select server

intents = discord.Intents.default()
intents.message_content = True   # ⚠️ Required for prefix commands
intents.guilds = True            # Guild join/leave events
intents.members = True           # ⚠️ Privileged — needs portal toggle

@bot.hybrid_command(name="ping", description="Pong!")
async def ping(ctx: commands.Context):
    await ctx.send(f"🏓 {bot.latency*1000:.0f}ms")

# Don't forget:
@bot.event
async def on_ready():
    await bot.tree.sync()  # ← registers slash commands with Discord

class MyCog(commands.Cog):
    def __init__(self, bot): self.bot = bot
    async def cog_load(self):   ...  # setup
    async def cog_unload(self): ...  # teardown

    @commands.hybrid_command()
    async def hello(self, ctx): await ctx.send("Hi!")

async def setup(bot):  # ← entry point
    await bot.add_cog(MyCog(bot))

from discord.ext import tasks

@tasks.loop(minutes=5)
async def heartbeat():
    ...  # periodic work

@heartbeat.before_loop
async def wait_ready():
    await bot.wait_until_ready()

heartbeat.start()  # call in on_ready or cog_load

class Confirm(discord.ui.View):
    value: bool | None = None

    @discord.ui.button(label="✅ Yes", style=discord.ButtonStyle.green)
    async def yes(self, interaction, button):
        self.value = True
        await interaction.response.edit_message(content="Confirmed!", view=None)
        self.stop()

    @discord.ui.button(label="❌ No", style=discord.ButtonStyle.red)
    async def no(self, interaction, button):
        self.value = False
        await interaction.response.edit_message(content="Cancelled.", view=None)
        self.stop()

# Usage:
view = Confirm(timeout=60)
await ctx.send("Proceed?", view=view)
await view.wait()

Need Discord integration?
│
├─ Just notifications? ──────────── 🪝 Tier 1 (Webhook)
│   └─ Drop index.ts, set env var, done.
│
├─ LLM needs to call Discord? ──── 🔧 Tier 2 (MCP)
│   └─ Write Python MCP server, add mcp.json, done.
│
└─ Full two-way bridge? ─────────── 🌉 Tier 3 (Extension + Bot)
    └─ OMP ext (.ts) + sidecar bot (.py) + systemd service.