telegram

We drive personal Telegram over MTProto with Telethon — this acts as the user's own account (a "userbot"), so unlike the Bot API it can read full history, list every conversation, and act on anyone the user can reach.

Credentials are injected as env vars by the connector:

• TELEGRAM_API_ID — app id
• TELEGRAM_API_HASH — app hash — secret, never echo
• TELEGRAM_SESSION_STRING — Telethon StringSession = full account access. Never log, echo, or print it. Treat it like the account password.

Setup — write the helper once per session

telethon is preinstalled in the sandbox. The helper is written to ./tg.py in the current working directory (the per-session workdir) — not a shared global path — so concurrent sessions never race.

Every state-changing command (send, reply, send-file, forward, edit, delete, react, mark-read) is gated: without a trailing --confirm it only DRY-RUNS (prints what it would do, changes nothing). Read commands run directly. --confirm is honored only as the last argument so a message/caption that merely contains "--confirm" can never silently confirm.

python3 -c "import telethon" 2>/dev/null || pip install --user --quiet telethon 2>/dev/null || true

cat > ./tg.py <<'PY'
import os, sys, json, asyncio
from telethon import TelegramClient
from telethon.sessions import StringSession
from telethon.tl import functions
from telethon.tl.types import ReactionEmoji

API_ID = int(os.environ["TELEGRAM_API_ID"])
API_HASH = os.environ["TELEGRAM_API_HASH"]
SESSION = os.environ["TELEGRAM_SESSION_STRING"]

_raw = sys.argv[1:]
# --confirm is only honored as the LAST token, and only one is stripped, so a
# message/caption that merely contains "--confirm" cannot silently confirm a write.
CONFIRM = bool(_raw) and _raw[-1] == "--confirm"
a = _raw[:-1] if CONFIRM else list(_raw)
cmd = a[0] if a else "help"
args = a[1:]
GATED = {"send", "reply", "send-file", "forward", "edit", "delete", "react", "mark-read"}


def out(o):
    print(json.dumps(o, ensure_ascii=False, default=str))


async def resolve(client, target):
    # 1) try direct resolve; 2) fall back to scanning dialogs by id or exact name
    #    (StringSession doesn't persist the entity cache, so a numeric id from a
    #    previous invocation may need the dialog scan to recover its access hash).
    for attempt in (lambda: client.get_entity(int(target)), lambda: client.get_entity(target)):
        try:
            return await attempt()
        except Exception:
            pass
    ti = None
    try:
        ti = int(target)
    except (ValueError, TypeError):
        pass
    async for d in client.iter_dialogs():
        if (ti is not None and d.id == ti) or d.name == target:
            return d.entity
    raise ValueError(f"could not resolve target: {target}")


def msg_row(m):
    return {"id": m.id, "date": str(m.date), "out": m.out, "sender_id": m.sender_id,
            "text": m.message, "media": bool(m.media)}


def need(n):
    if len(args) < n:
        raise ValueError(f"{cmd} needs {n} argument(s), got {len(args)}")


async def run():
    if cmd in GATED and not CONFIRM:
        out({"dry_run": True, "command": cmd, "args": args,
             "note": "re-run with --confirm as the LAST argument to actually perform this write"})
        return
    async with TelegramClient(StringSession(SESSION), API_ID, API_HASH) as cl:
        if cmd == "whoami":
            me = await cl.get_me()
            out({"id": me.id, "username": me.username,
                 "name": ((me.first_name or "") + " " + (me.last_name or "")).strip(), "phone": me.phone})

        elif cmd == "list-chats":
            limit = int(args[0]) if args and args[0].lstrip("-").isdigit() else 20
            unread_only = "unread-only" in args
            res = []
            async for d in cl.iter_dialogs(limit=limit):
                if unread_only and not d.unread_count:
                    continue
                res.append({"name": d.name, "id": d.id, "group": d.is_group,
                            "channel": d.is_channel, "user": d.is_user, "unread": d.unread_count})
            out(res)

        elif cmd == "unread":
            res = []
            async for d in cl.iter_dialogs():
                if d.unread_count:
                    res.append({"name": d.name, "id": d.id, "unread": d.unread_count,
                                "group": d.is_group, "channel": d.is_channel})
            out(sorted(res, key=lambda x: -x["unread"]))

        elif cmd == "get-messages":
            need(1); ent = await resolve(cl, args[0])
            n = int(args[1]) if len(args) > 1 else 50
            rows = [msg_row(m) async for m in cl.iter_messages(ent, limit=n)]
            rows.reverse()
            out(rows)

        elif cmd == "search":
            need(2); ent = await resolve(cl, args[0])
            q = args[1]; n = int(args[2]) if len(args) > 2 else 30
            out([msg_row(m) async for m in cl.iter_messages(ent, search=q, limit=n)])

        elif cmd == "search-global":
            need(1); q = args[0]; n = int(args[1]) if len(args) > 1 else 30
            rows = []
            async for m in cl.iter_messages(None, search=q, limit=n):
                r = msg_row(m); r["chat_id"] = m.chat_id
                rows.append(r)
            out(rows)

        elif cmd == "contacts":
            res = await cl(functions.contacts.GetContactsRequest(hash=0))
            out([{"id": u.id, "username": u.username,
                  "name": ((u.first_name or "") + " " + (u.last_name or "")).strip(), "phone": u.phone}
                 for u in res.users])

        elif cmd == "chat-info":
            need(1); ent = await resolve(cl, args[0])
            info = {"id": ent.id, "type": type(ent).__name__,
                    "title": getattr(ent, "title", None),
                    "name": ((getattr(ent, "first_name", "") or "") + " " + (getattr(ent, "last_name", "") or "")).strip() or None,
                    "username": getattr(ent, "username", None)}
            try:
                info["participants"] = (await cl.get_participants(ent, limit=1)).total
            except Exception:
                pass
            out(info)

        elif cmd == "message-link":
            need(2); ent = await resolve(cl, args[0]); mid = int(args[1])
            try:
                r = await cl(functions.channels.ExportMessageLinkRequest(channel=ent, id=mid))
                out({"link": r.link})
            except Exception as e:
                out({"error": f"links only available for channels/supergroups: {e}"})

        elif cmd == "download-media":
            need(2); ent = await resolve(cl, args[0]); mid = int(args[1])
            outdir = args[2] if len(args) > 2 else "./tg_downloads"
            os.makedirs(outdir, exist_ok=True)
            m = await cl.get_messages(ent, ids=mid)
            if not m or not m.media:
                out({"error": "no media on that message"}); return
            path = await cl.download_media(m, file=outdir)
            out({"downloaded": path})

        # ---- gated writes (need trailing --confirm) ----
        elif cmd == "send":
            need(2); ent = await resolve(cl, args[0])
            m = await cl.send_message(ent, args[1])
            out({"sent": True, "id": m.id})

        elif cmd == "reply":
            need(3); ent = await resolve(cl, args[0])
            m = await cl.send_message(ent, args[2], reply_to=int(args[1]))
            out({"sent": True, "id": m.id, "reply_to": int(args[1])})

        elif cmd == "send-file":
            need(2); ent = await resolve(cl, args[0])
            caption = args[2] if len(args) > 2 else None
            m = await cl.send_file(ent, args[1], caption=caption)
            out({"sent": True, "id": m.id})

        elif cmd == "forward":
            need(3); src = await resolve(cl, args[0]); mid = int(args[1]); dst = await resolve(cl, args[2])
            fwd = await cl.forward_messages(dst, mid, src)
            out({"forwarded": True, "id": getattr(fwd, "id", None) or [x.id for x in fwd]})

        elif cmd == "edit":
            need(3); ent = await resolve(cl, args[0])
            m = await cl.edit_message(ent, int(args[1]), args[2])
            out({"edited": True, "id": m.id})

        elif cmd == "delete":
            need(2); ent = await resolve(cl, args[0])
            await cl.delete_messages(ent, int(args[1]))
            out({"deleted": True, "id": int(args[1])})

        elif cmd == "react":
            need(3); ent = await resolve(cl, args[0]); mid = int(args[1]); emoji = args[2]
            await cl(functions.messages.SendReactionRequest(
                peer=ent, msg_id=mid, reaction=[ReactionEmoji(emoticon=emoji)]))
            out({"reacted": True, "id": mid, "emoji": emoji})

        elif cmd == "mark-read":
            need(1); ent = await resolve(cl, args[0])
            await cl.send_read_acknowledge(ent)
            out({"marked_read": True})

        else:
            out({"error": f"unknown command: {cmd}"}); sys.exit(1)


async def main():
    try:
        await run()
    except SystemExit:
        raise
    except Exception as e:
        out({"error": f"{type(e).__name__}: {e}"})
        sys.exit(1)


asyncio.run(main())
PY
echo "helper ready"

Verify the connection first

python3 ./tg.py whoami
# → {"id": 8367450178, "username": "GermeyAce", "name": "Germey", "phone": "..."}

On an auth/session error the stored session is dead — tell the user to reconnect at https://auth.acedata.cloud/user/connections.

Read recipes

Goal	Command
Recent conversations	python3 ./tg.py list-chats 20
Only chats with unread (ranked)	python3 ./tg.py unread
A chat's history (oldest→newest)	python3 ./tg.py get-messages <target> 50
Search inside one chat	python3 ./tg.py search <target> "kw" 30
Search across ALL chats	python3 ./tg.py search-global "kw" 30
List contacts	python3 ./tg.py contacts
Info about a chat/user	python3 ./tg.py chat-info <target>
t.me link to a message	python3 ./tg.py message-link <target> <msg_id>

<target> = numeric id (most reliable — from list-chats), @username, phone, or exact chat name. In message rows, out:true = sent by the user; media:true = has an attachment.

Summarize-unread pattern: unread → pick the chats that matter → get-messages <id> N on each → summarize. Don't dump 20k messages; sample the most-unread / most-relevant.

Media

# Download an attachment from a message → returns the saved path
python3 ./tg.py download-media <target> <msg_id> ./tg_downloads
# Send a local file or a URL (optional caption) — GATED
python3 ./tg.py send-file <target> /path/or/https-url "caption" --confirm

To hand a downloaded file back to the user as a link, upload it to the CDN (see the cos-upload skill) after download-media.

Write recipes — all GATED (dry-run unless trailing --confirm)

Sending/editing/deleting acts as the real user. Always run the dry run first, show the user exactly what will happen, get an explicit "yes", then re-run with --confirm as the last argument. Never bulk-send.

python3 ./tg.py send    <target> "text"                          # → dry_run; add --confirm to send
python3 ./tg.py reply   <target> <msg_id> "text" --confirm
python3 ./tg.py forward <from_target> <msg_id> <to_target> --confirm
python3 ./tg.py edit    <target> <msg_id> "new text" --confirm   # own messages
python3 ./tg.py delete  <target> <msg_id> --confirm              # destructive
python3 ./tg.py react   <target> <msg_id> "👍" --confirm
python3 ./tg.py mark-read <target> --confirm                     # sends read receipts

The dry run returns {"dry_run": true, "command": ..., "args": [...]} — present that to the user verbatim as the confirmation prompt.

Gotchas — surface before the user is surprised

• This is the user's real account. Confirm before any write; reading exposes private chats.
• FloodWaitError: Telegram rate-limits userbots. On a flood-wait of N seconds, tell the user to retry after N — never loop/retry aggressively (escalates toward a ban).
• Dead session: revoked from Telegram → Settings → Devices, or ~6-month inactivity. On AuthKeyError/unauthorized, reconnect the connector (don't retry).
• Never print TELEGRAM_SESSION_STRING / TELEGRAM_API_HASH — full-account secrets.
• Targets: prefer the numeric id from list-chats (the helper recovers its access hash by scanning dialogs); names need an exact match, usernames need a leading @.
• message-link only works for public channels/supergroups; private 1:1 / basic groups return an error (no shareable link exists).
• edit/delete generally only apply to the user's own messages (admins can delete others in groups they manage).