Agent Skills
› NeverSight/learn-skills.dev
› telegram
telegram
GitHub基于Telethon的个人Telegram控制技能，支持聊天管理、消息读写、媒体下载及文件发送等。需配置API凭证，通过tg.py脚本执行，写操作默认Dry-Run，加--confirm生效，确保账户安全与操作可控。
Trigger Scenarios
用户提及 Telegram
查询 Telegram 聊天/群组/联系人
阅读或转发 Telegram 消息
查看未读消息
在 Telegram 上发送文件或消息
Install
npx skills add NeverSight/learn-skills.dev --skill telegram -g -y
SKILL.md
Frontmatter
{
"name": "telegram",
"license": "Apache-2.0",
"metadata": {
"author": "acedatacloud",
"version": "1.1"
},
"connections": [
"telegram"
],
"description": "Full personal Telegram control over MTProto (Telethon) with the user's own account — list\/search chats, read & summarize history, see unread, look up contacts & chat info, download media, and send \/ reply \/ forward \/ edit \/ delete \/ react \/ send files \/ mark read. Use when the user mentions Telegram, a Telegram chat\/group\/contact, \"我的 Telegram\", reading\/replying\/forwarding\/summarizing Telegram messages, their unread Telegram, or sending a file\/message on Telegram.",
"when_to_use": "Trigger for anything on the user's personal Telegram account: list recent\nconversations or just the unread ones, read \/ summarize a chat or group,\nsearch one chat or across all chats, look up a contact or a chat's info,\ndownload a photo\/file from a message, or take an action — send, reply,\nforward, edit, delete, react, send a file, or mark a chat read. This drives\nthe user's OWN account over MTProto (not a bot), so it sees everything they see.\n",
"allowed_tools": [
"Bash"
]
}
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— TelethonStringSession= 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
idfromlist-chats(the helper recovers its access hash by scanning dialogs); names need an exact match, usernames need a leading@. message-linkonly works for public channels/supergroups; private 1:1 / basic groups return an error (no shareable link exists).edit/deletegenerally only apply to the user's own messages (admins can delete others in groups they manage).
Version History
- e0220ca Current 2026-07-05 22:55
