Build your first app

Markdown

In about ten minutes you will post a message from your terminal, then run a small bot that answers people. You need to be allowed to create apps in a workspace: an owner or admin always is, and a member is when the workspace lets members make apps (Manage → Settings → Who can create apps). When it works by requests, ask for the app and generate its token once an admin approves it.

Create the app

Open Manage → Installed apps → Create new app, name it, and choose these permissions: chat:write, channels:read and channels:join. Copy the bot token when it is shown. It appears once; you can save it in your Vault from the same dialog, or replace it later.
Terminal window
```
export BASE="https://amedeus.app/api/public/apps/v1"
export TOKEN="amd_bot_…"
```
Find a channel and join it

An app is created in no channel: it joins the ones it needs.
Terminal window
curl "$BASE/channels" -H "Authorization: Bearer $TOKEN"
const response = await fetch(`${process.env.BASE}/channels`, { headers: { Authorization: `Bearer ${process.env.TOKEN}` }, }); const { channels } = await response.json();
import json, os, urllib.request request = urllib.request.Request( f"{os.environ['BASE']}/channels", headers={"Authorization": f"Bearer {os.environ['TOKEN']}", "User-Agent": "my-app/1.0"}, ) channels = json.load(urllib.request.urlopen(request))["channels"]
Every public channel comes back with its id and is_member. Join one:
Terminal window
```
curl -X POST "$BASE/channels/<channel id>/join" -H "Authorization: Bearer $TOKEN"
```

Post a message

curl -X POST "$BASE/messages" \
  -H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
  -d '{"channel": "<channel id>", "text": "Hello from my first app :wave:"}'

await fetch(`${process.env.BASE}/messages`, {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.TOKEN}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ channel: "<channel id>", text: "Hello from my first app :wave:" }),
});

import json, os, urllib.request

request = urllib.request.Request(
    f"{os.environ['BASE']}/messages",
    data=json.dumps({"channel": "<channel id>", "text": "Hello from my first app :wave:"}).encode(),
    headers={
        "Authorization": f"Bearer {os.environ['TOKEN']}",
        "Content-Type": "application/json",
        "User-Agent": "my-app/1.0",
    },
    method="POST",
)
urllib.request.urlopen(request).read()

The message appears in the channel from your app, with its name and an APP badge, and the response contains it. See Post a message.

Let people write to it

Under App Home, turn on Allow people to message this app. Under Event subscriptions, you will set a URL in the next step: saving it adds the permissions the events need.

Answer people

To hear what people write, your app needs a public HTTPS address that Amedeus can call. Under Event subscriptions, set the Request URL, turn on message.im and app_mention, copy the signing secret, and save. Amedeus checks your URL with a challenge first (see Receive events).

A bot that repeats what it hears, with no dependencies:

JavaScript
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}`));

Run it with SIGNING_SECRET=… BOT_TOKEN=… node bot.mjs (Node 22).

"""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()

Run it with SIGNING_SECRET=… BOT_TOKEN=… python bot.py (Python 3.10 or later).

Publish it (a tunnel works while you build), save its URL, and write to the app: it answers. Messages written by apps never reach apps, so your bot cannot trigger itself.

This bot is also a ready-made project, with a Dockerfile and a .env.example: Templates.

Let people ask for things with slash commands.
Read the event payloads and the retry rules.
Handle errors by their error code.

Build your first app

Next