Full personal Telegram account control via MTProto (Telethon) — list/search chats, read/summarize history, see unread, look up contacts, download media, and send/reply/forward/edit/delete/react/send files/mark read. Use when the user mentions Telegram or their own Telegram account.
How this skill is triggered — by the user, by Claude, or both
Slash command
/acedatacloud-ai-media:telegramWhen to use
Trigger for anything on the user's personal Telegram account: list recent conversations or just the unread ones, read / summarize a chat or group, search one chat or across all chats, look up a contact or a chat's info, download a photo/file from a message, or take an action — send, reply, forward, edit, delete, react, send a file, or mark a chat read. This drives the user's OWN account over MTProto (not a bot), so it sees everything they see.
The summary Claude sees in its skill listing — used to decide when to auto-load this skill
We drive **personal** Telegram over MTProto with [Telethon](https://docs.telethon.dev/) —
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 idTELEGRAM_API_HASH — app hash — secret, never echoTELEGRAM_SESSION_STRING — Telethon StringSession = full account access. Never log,
echo, or print it. Treat it like the account password.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"
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.
| 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.
# 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.
--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.
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).AuthKeyError/unauthorized, reconnect the connector (don't retry).TELEGRAM_SESSION_STRING / TELEGRAM_API_HASH — full-account secrets.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).npx claudepluginhub acedatacloud/skills --plugin acedatacloud-ai-toolsChecks inbound Telegram messages and replies using tg-tools. Supports contact lookup, group/channel scraping, and export with last messages.
Downloads, exports, converts, and filters Telegram chats via the telegram-download-chat CLI. Supports message download, JSON-to-TXT/HTML/PDF conversion, date/user/keyword filtering, media download, and subchat extraction.
Provides CLI for Telegram Bot API using bot tokens: authenticate, send messages, read chats, get info, manage multi-bot setups. For CI/CD and server-side messaging automation.