Templates

Markdown

Each template is a small server with no dependencies: it verifies the signature of every request, answers within 3 seconds, and does its work afterwards. They are the code the guides describe, and the same files are tested against a real workspace.

Echo bot

Repeats what people write to it, in the thread of the message. Start here for anything that answers messages or mentions.

Slash command bot

Answers /deploy staging: checks who may run it, starts the work, replies privately and tells the channel.

Echo bot

Needs chat:write, the message.im and app_mention events, and Allow people to message this app under App Home. See the quickstart for the setup.

Node
Python

// An Amedeus app that answers when someone writes to it or mentions it. Node 22, no dependencies.
//
//   SIGNING_SECRET=… BOT_TOKEN=… node bot.mjs
//
// Amedeus calls this server for every event the app subscribed to. Verify the signature, answer 200
// within 3 seconds, and do the work afterwards.
import crypto from "node:crypto";
import http from "node:http";

const { SIGNING_SECRET, BOT_TOKEN, PORT = "3000" } = process.env;
const API = process.env.AMEDEUS_API ?? "https://amedeus.app/api/public/apps/v1";

if (!SIGNING_SECRET || !BOT_TOKEN) {
  console.error("Set SIGNING_SECRET and BOT_TOKEN (see .env.example).");
  process.exit(1);
}

function isFromAmedeus(headers, rawBody) {
  const timestamp = headers["x-amedeus-request-timestamp"];
  // Refuse old requests, so a captured one can't be replayed.
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
  const expected =
    "v0=" +
    crypto.createHmac("sha256", SIGNING_SECRET).update(`v0:${timestamp}:${rawBody}`).digest("hex");
  const given = String(headers["x-amedeus-signature"] ?? "");
  return (
    given.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(given), Buffer.from(expected))
  );
}

// One message can arrive as more than one event (a mention in a channel is both `message.channel`
// and `app_mention`), and a retry repeats an event. Answer each message once.
const answered = new Set();
function firstTime(messageId) {
  if (answered.has(messageId)) return false;
  answered.add(messageId);
  // Keep the memory bounded: forget the oldest once there are a thousand.
  if (answered.size > 1000) answered.delete(answered.values().next().value);
  return true;
}

async function reply(event) {
  const response = await fetch(`${API}/messages`, {
    method: "POST",
    headers: { Authorization: `Bearer ${BOT_TOKEN}`, "Content-Type": "application/json" },
    body: JSON.stringify({
      channel: event.channel,
      // In a channel, answer in the thread of the message; in a direct message, just answer.
      parent_id: event.channel_kind === "dm" ? undefined : (event.parent_id ?? event.message_id),
      text: `You said: ${event.text}`,
    }),
  });
  // Every error has a stable `error` code.
  if (!response.ok) console.error("Could not reply:", await response.json());
}

http
  .createServer((req, res) => {
    let rawBody = "";
    req.on("data", (chunk) => (rawBody += chunk));
    req.on("end", () => {
      if (!isFromAmedeus(req.headers, rawBody)) return res.writeHead(401).end();
      const payload = JSON.parse(rawBody);
      // Amedeus checks your URL with a challenge when you save it.
      if (payload.type === "url_verification") return res.writeHead(200).end(payload.challenge);
      // Answer within 3 seconds, then do the work.
      res.writeHead(200).end("ok");
      if (payload.type === "event_callback" && firstTime(payload.event.message_id))
        reply(payload.event).catch((error) => console.error("Could not reply:", error));
    });
  })
  .listen(Number(PORT), () => console.log(`Listening on :${PORT}`));

Download the project (bot.mjs, package.json, .env.example, Dockerfile, README.md):

for f in bot.mjs package.json .env.example Dockerfile README.md; do
  curl -O "https://docs.amedeus.app/templates/echo-bot-node/$f"
done

"""An Amedeus app that answers when someone writes to it or mentions it.

Python 3.10+, standard library only.

    SIGNING_SECRET=… BOT_TOKEN=… python bot.py

Amedeus calls this server for every event the app subscribed to. Verify the signature, answer 200
within 3 seconds, and do the work afterwards.
"""

import hashlib
import hmac
import json
import os
import socket
import threading
import time
import urllib.error
import urllib.request
from collections import OrderedDict
from http.server import BaseHTTPRequestHandler, HTTPServer

SIGNING_SECRET = os.environ.get("SIGNING_SECRET", "").encode()
BOT_TOKEN = os.environ.get("BOT_TOKEN", "")
API = os.environ.get("AMEDEUS_API", "https://amedeus.app/api/public/apps/v1")
PORT = int(os.environ.get("PORT", "3000"))

if not SIGNING_SECRET or not BOT_TOKEN:
    raise SystemExit("Set SIGNING_SECRET and BOT_TOKEN (see .env.example).")


def is_from_amedeus(headers, raw_body: bytes) -> bool:
    timestamp = headers.get("X-Amedeus-Request-Timestamp", "")
    # Refuse old requests, so a captured one can't be replayed.
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        return False
    signed = f"v0:{timestamp}:".encode() + raw_body
    expected = "v0=" + hmac.new(SIGNING_SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, headers.get("X-Amedeus-Signature", ""))


# One message can arrive as more than one event (a mention in a channel is both `message.channel`
# and `app_mention`), and a retry repeats an event. Answer each message once.
answered: OrderedDict[str, bool] = OrderedDict()
answered_lock = threading.Lock()


def first_time(message_id: str) -> bool:
    with answered_lock:
        if message_id in answered:
            return False
        answered[message_id] = True
        # Keep the memory bounded: forget the oldest once there are a thousand.
        if len(answered) > 1000:
            answered.popitem(last=False)
        return True


def reply(event: dict) -> None:
    body = {"channel": event["channel"], "text": f"You said: {event['text']}"}
    # In a channel, answer in the thread of the message; in a direct message, just answer.
    if event["channel_kind"] != "dm":
        body["parent_id"] = event.get("parent_id") or event["message_id"]
    request = urllib.request.Request(
        f"{API}/messages",
        data=json.dumps(body).encode(),
        headers={
            "Authorization": f"Bearer {BOT_TOKEN}",
            "Content-Type": "application/json",
            # Name your app: some proxies refuse urllib's default User-Agent (Cloudflare answers 1010).
            "User-Agent": "amedeus-echo-bot/1.0",
        },
        method="POST",
    )
    try:
        urllib.request.urlopen(request, timeout=10).read()
    except urllib.error.HTTPError as error:
        # Every error has a stable `error` code.
        print("Could not reply:", error.read().decode())
    except OSError as error:
        print("Could not reply:", error)


class DualStackServer(HTTPServer):
    """Listen on IPv4 and IPv6, so `localhost` works whichever one a tunnel or proxy picks."""

    address_family = socket.AF_INET6

    def server_bind(self) -> None:
        self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
        super().server_bind()


def make_server() -> HTTPServer:
    try:
        return DualStackServer(("::", PORT), Handler)
    except OSError:  # no IPv6 on this host
        return HTTPServer(("0.0.0.0", PORT), Handler)


class Handler(BaseHTTPRequestHandler):
    def _send(self, status: int, body: bytes = b"") -> None:
        self.send_response(status)
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self) -> None:
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not is_from_amedeus(self.headers, raw_body):
            return self._send(401)
        payload = json.loads(raw_body)
        # Amedeus checks your URL with a challenge when you save it.
        if payload["type"] == "url_verification":
            return self._send(200, payload["challenge"].encode())
        # Answer within 3 seconds, then do the work.
        self._send(200, b"ok")
        if payload["type"] == "event_callback" and first_time(payload["event"]["message_id"]):
            threading.Thread(target=reply, args=(payload["event"],)).start()


if __name__ == "__main__":
    print(f"Listening on :{PORT}")
    make_server().serve_forever()

Download the project (bot.py, .env.example, Dockerfile, README.md):

for f in bot.py .env.example Dockerfile README.md; do
  curl -O "https://docs.amedeus.app/templates/echo-bot-python/$f"
done

Slash command bot

Needs the commands permission and a command registered under Slash commands (/deploy). The bot allows only the user ids you list, answers refusals and usage hints privately, and tells the channel when a deploy starts. Without DEPLOY_WEBHOOK_URL it is a dry run, so you can try it safely.

Node
Python

// An Amedeus app that answers a slash command: `/deploy staging`. Node 22, no dependencies.
//
//   SIGNING_SECRET=… ALLOWED_USER_IDS=<your user id> node bot.mjs
//
// Amedeus sends the command here, signed. Verify it, answer 200 within 3 seconds, then reply through
// the command's `response_url`: privately to the person who ran it, or to the whole channel.
import crypto from "node:crypto";
import http from "node:http";

const { SIGNING_SECRET, PORT = "3000", DEPLOY_WEBHOOK_URL } = process.env;
// Who may deploy: a comma-separated list of Amedeus user ids. Everyone else gets a private refusal.
const ALLOWED_USER_IDS = (process.env.ALLOWED_USER_IDS ?? "").split(",").filter(Boolean);
const ENVIRONMENTS = (process.env.ENVIRONMENTS ?? "staging,production").split(",");

if (!SIGNING_SECRET) {
  console.error("Set SIGNING_SECRET (see .env.example).");
  process.exit(1);
}

function isFromAmedeus(headers, rawBody) {
  const timestamp = headers["x-amedeus-request-timestamp"];
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false;
  const expected =
    "v0=" +
    crypto.createHmac("sha256", SIGNING_SECRET).update(`v0:${timestamp}:${rawBody}`).digest("hex");
  const given = String(headers["x-amedeus-signature"] ?? "");
  return (
    given.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(given), Buffer.from(expected))
  );
}

// The response URL works for 30 minutes and 5 times. `ephemeral` (the default) is read only by the
// person who ran the command; `in_channel` is a message from the app for everyone.
async function respond(command, text, responseType = "ephemeral") {
  const response = await fetch(command.response_url, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text, response_type: responseType }),
  });
  if (!response.ok) console.error("Could not respond:", await response.json());
}

async function handleCommand(command) {
  if (!ALLOWED_USER_IDS.includes(command.user))
    return respond(command, "You're not allowed to deploy.");

  const environment = command.text.trim();
  if (!ENVIRONMENTS.includes(environment))
    return respond(command, `Usage: \`/deploy <${ENVIRONMENTS.join(" | ")}>\``);

  // Start the deploy: here, a POST to the URL your platform gives you. Without one it is a dry run.
  if (DEPLOY_WEBHOOK_URL) await fetch(DEPLOY_WEBHOOK_URL, { method: "POST" });
  console.log(`Deploy to ${environment} requested by ${command.user}`);

  await respond(command, `**Deploy to ${environment} started.** I'll post here when it finishes.`);
  await respond(command, `A deploy to **${environment}** was started.`, "in_channel");
}

http
  .createServer((req, res) => {
    let rawBody = "";
    req.on("data", (chunk) => (rawBody += chunk));
    req.on("end", () => {
      if (!isFromAmedeus(req.headers, rawBody)) return res.writeHead(401).end();
      const payload = JSON.parse(rawBody);
      if (payload.type === "url_verification") return res.writeHead(200).end(payload.challenge);
      // Answer within 3 seconds, then do the work.
      res.writeHead(200).end("ok");
      if (payload.type === "slash_command" && payload.command === "/deploy")
        handleCommand(payload).catch((error) => console.error("Command failed:", error));
    });
  })
  .listen(Number(PORT), () => console.log(`Listening on :${PORT}`));

for f in bot.mjs package.json .env.example Dockerfile README.md; do
  curl -O "https://docs.amedeus.app/templates/slash-command-node/$f"
done

"""An Amedeus app that answers a slash command: `/deploy staging`.

Python 3.10+, standard library only.

    SIGNING_SECRET=… ALLOWED_USER_IDS=<your user id> python bot.py

Amedeus sends the command here, signed. Verify it, answer 200 within 3 seconds, then reply through the
command's `response_url`: privately to the person who ran it, or to the whole channel.
"""

import hashlib
import hmac
import json
import os
import socket
import threading
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

SIGNING_SECRET = os.environ.get("SIGNING_SECRET", "").encode()
PORT = int(os.environ.get("PORT", "3000"))
DEPLOY_WEBHOOK_URL = os.environ.get("DEPLOY_WEBHOOK_URL")
# Who may deploy: a comma-separated list of Amedeus user ids. Everyone else gets a private refusal.
ALLOWED_USER_IDS = [u for u in os.environ.get("ALLOWED_USER_IDS", "").split(",") if u]
ENVIRONMENTS = os.environ.get("ENVIRONMENTS", "staging,production").split(",")

if not SIGNING_SECRET:
    raise SystemExit("Set SIGNING_SECRET (see .env.example).")


def is_from_amedeus(headers, raw_body: bytes) -> bool:
    timestamp = headers.get("X-Amedeus-Request-Timestamp", "")
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        return False
    signed = f"v0:{timestamp}:".encode() + raw_body
    expected = "v0=" + hmac.new(SIGNING_SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, headers.get("X-Amedeus-Signature", ""))


def post_json(url: str, body: dict | None = None) -> None:
    request = urllib.request.Request(
        url,
        data=json.dumps(body or {}).encode(),
        headers={
            "Content-Type": "application/json",
            # Name your app: some proxies refuse urllib's default User-Agent (Cloudflare answers 1010).
            "User-Agent": "amedeus-slash-command-bot/1.0",
        },
        method="POST",
    )
    try:
        urllib.request.urlopen(request, timeout=10).read()
    except urllib.error.HTTPError as error:
        print("Request failed:", error.read().decode())
    except OSError as error:
        print("Request failed:", error)


# The response URL works for 30 minutes and 5 times. "ephemeral" (the default) is read only by the
# person who ran the command; "in_channel" is a message from the app for everyone.
def respond(command: dict, text: str, response_type: str = "ephemeral") -> None:
    post_json(command["response_url"], {"text": text, "response_type": response_type})


def handle_command(command: dict) -> None:
    if command["user"] not in ALLOWED_USER_IDS:
        return respond(command, "You're not allowed to deploy.")

    environment = command["text"].strip()
    if environment not in ENVIRONMENTS:
        return respond(command, f"Usage: `/deploy <{' | '.join(ENVIRONMENTS)}>`")

    # Start the deploy: here, a POST to the URL your platform gives you. Without one it is a dry run.
    if DEPLOY_WEBHOOK_URL:
        post_json(DEPLOY_WEBHOOK_URL)
    print(f"Deploy to {environment} requested by {command['user']}")

    respond(command, f"**Deploy to {environment} started.** I'll post here when it finishes.")
    respond(command, f"A deploy to **{environment}** was started.", "in_channel")


class DualStackServer(HTTPServer):
    """Listen on IPv4 and IPv6, so `localhost` works whichever one a tunnel or proxy picks."""

    address_family = socket.AF_INET6

    def server_bind(self) -> None:
        self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
        super().server_bind()


def make_server() -> HTTPServer:
    try:
        return DualStackServer(("::", PORT), Handler)
    except OSError:  # no IPv6 on this host
        return HTTPServer(("0.0.0.0", PORT), Handler)


class Handler(BaseHTTPRequestHandler):
    def _send(self, status: int, body: bytes = b"") -> None:
        self.send_response(status)
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self) -> None:
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        if not is_from_amedeus(self.headers, raw_body):
            return self._send(401)
        payload = json.loads(raw_body)
        if payload["type"] == "url_verification":
            return self._send(200, payload["challenge"].encode())
        # Answer within 3 seconds, then do the work.
        self._send(200, b"ok")
        if payload["type"] == "slash_command" and payload["command"] == "/deploy":
            threading.Thread(target=handle_command, args=(payload,)).start()


if __name__ == "__main__":
    print(f"Listening on :{PORT}")
    make_server().serve_forever()

for f in bot.py .env.example Dockerfile README.md; do
  curl -O "https://docs.amedeus.app/templates/slash-command-python/$f"
done

Deploying a template

Each folder has a Dockerfile: build it, set the variables of .env.example, and give the container a public HTTPS address. Any host works (a VPS, Dokploy, a serverless container).

If you call the Amedeus API from Python behind Cloudflare, send your own User-Agent, as the templates do: Cloudflare refuses the default one of urllib (error 1010).