# Build your first app

> Create an app, post your first message, then make it answer when someone writes to it.

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.

1. **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.

```sh
export BASE="https://amedeus.app/api/public/apps/v1"
export TOKEN="amd_bot_…"
```

2. **Find a channel and join it**

An app is created in no channel: it joins the ones it needs.

**curl**

```sh
curl "$BASE/channels" -H "Authorization: Bearer $TOKEN"
```

**JavaScript**

```js
const response = await fetch(`${process.env.BASE}/channels`, {
  headers: { Authorization: `Bearer ${process.env.TOKEN}` },
});
const { channels } = await response.json();
```

**Python**

```python
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:

```sh
curl -X POST "$BASE/channels/<channel id>/join" -H "Authorization: Bearer $TOKEN"
```

3. **Post a message**

**curl**

```sh
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:"}'
```

**JavaScript**

```js
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:" }),
});
```

**Python**

```python
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](/api-reference/operations/post-message/).

4. **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.

5. **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](/guides/events/)).

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

**JavaScript**

**bot.mjs**

```js
// 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).

**Python**

**bot.py**

```python
"""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](/guides/templates/).

## Next

- Let people ask for things with [slash commands](/guides/slash-commands/).
- Read the [event payloads](/guides/events/) and the [retry rules](/guides/events/#retries).
- Handle [errors](/reference/errors/) by their `error` code.