Vladimir Buzalka
128 lines
5.0 KiB
Python
128 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
MCP server pro Telegram — FastMCP
|
|
Spustit: python mcp_telegram.py
|
|
|
|
Dává agentům nástroje pro komunikaci s Vladem přes Telegram.
|
|
|
|
Nástroje:
|
|
notifikace — pošle Vladovi zprávu přes bota (oznámení, nečeká na odpověď)
|
|
zeptej_se — pošle Vladovi otázku přes bota a POČKÁ na jeho odpověď
|
|
posli_jako_agent — pošle Vladovi zprávu z účtu agenta "Claude Buzalka" (persona)
|
|
zeptej_se_jako_agent — otázka z účtu agenta + čekání na odpověď
|
|
|
|
Dvě cesty:
|
|
• BOT (@Vlado_Claude_Bot) — bezpečné, bez rizika banu; doporučené pro notifikace/dotazy.
|
|
• USER účet (Claude Buzalka) — vystupuje jako "osoba"; má riziko banu (nový účet).
|
|
|
|
POZOR (bot): getUpdates smí pollovat jen JEDEN proces. Když běží víc agentů, kteří
|
|
naráz čekají na odpověď přes bota, kradou si zprávy. Pro souběžné dotazy víc agentů
|
|
použij user účet (každý agent vlastní session) — viz Knihovny/telegram_user.py.
|
|
"""
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
# Kořen projektu na sys.path, ať jdou importovat sdílené knihovny
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(BASE_DIR))
|
|
|
|
from Knihovny.telegram_notify import posli_telegram, zeptej_se_telegram
|
|
from Knihovny.telegram_user import posli_jako_ja, zeptej_se_jako
|
|
|
|
|
|
def log(msg: str):
|
|
print(msg, file=sys.stderr, flush=True)
|
|
|
|
|
|
def _v_threadu(fn, *args, **kwargs):
|
|
"""Spustí synchronní (telethon.sync) funkci ve vlastním vlákně s čistou
|
|
event-loop, aby nekolidovala s běžící async smyčkou FastMCP."""
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
try:
|
|
return fn(*args, **kwargs)
|
|
finally:
|
|
loop.close()
|
|
asyncio.set_event_loop(None)
|
|
|
|
|
|
# Vladův hlavní Telegram účet (user id == chat_id u bota)
|
|
VLADO_UID = 6639316354
|
|
|
|
mcp = FastMCP("telegram")
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# BOT — bezpečná cesta
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
@mcp.tool()
|
|
def notifikace(text: str, tise: bool = False) -> dict:
|
|
"""Pošle Vladovi oznámení přes Telegram bota (nečeká na odpověď).
|
|
|
|
Použij pro informování o průběhu, dokončení úlohy, chybě apod.
|
|
|
|
:param text: text zprávy
|
|
:param tise: True = tichá zpráva bez zvuku/upozornění
|
|
:return: {"ok": True}
|
|
"""
|
|
posli_telegram(text, disable_notification=tise)
|
|
return {"ok": True}
|
|
|
|
|
|
@mcp.tool()
|
|
def zeptej_se(otazka: str, timeout_s: int = 180) -> dict:
|
|
"""Pošle Vladovi otázku přes bota a POČKÁ na jeho textovou odpověď.
|
|
|
|
Vhodné, když agent během běhu potřebuje rozhodnutí. Blokuje až `timeout_s` sekund.
|
|
|
|
:param otazka: text otázky (klidně s nabídkou možností, např. "(ano/ne)")
|
|
:param timeout_s: jak dlouho čekat na odpověď
|
|
:return: {"odpoved": <text>} nebo {"odpoved": None, "timeout": True}
|
|
"""
|
|
odp = zeptej_se_telegram(otazka, timeout=timeout_s)
|
|
if odp is None:
|
|
return {"odpoved": None, "timeout": True}
|
|
return {"odpoved": odp}
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# USER účet "Claude Buzalka" — persona (riziko banu)
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
@mcp.tool()
|
|
async def posli_jako_agent(text: str) -> dict:
|
|
"""Pošle Vladovi zprávu z účtu agenta "Claude Buzalka" (ne z bota).
|
|
|
|
:param text: text zprávy
|
|
:return: {"ok": True}
|
|
"""
|
|
await asyncio.to_thread(_v_threadu, posli_jako_ja, VLADO_UID, text)
|
|
return {"ok": True}
|
|
|
|
|
|
@mcp.tool()
|
|
async def zeptej_se_jako_agent(otazka: str, agent: str = "Claude", timeout_s: int = 180) -> dict:
|
|
"""Pošle Vladovi otázku z účtu agenta a POČKÁ na odpověď (stačí běžná zpráva).
|
|
|
|
:param otazka: text otázky
|
|
:param agent: jméno agenta (štítek + session); výchozí "Claude"
|
|
:param timeout_s: jak dlouho čekat
|
|
:return: {"odpoved": <text>} nebo {"odpoved": None, "timeout": True}
|
|
"""
|
|
odp = await asyncio.to_thread(
|
|
_v_threadu, zeptej_se_jako, agent, otazka,
|
|
komu=VLADO_UID, session=None, timeout=timeout_s, vyzaduj_reply=False,
|
|
)
|
|
if odp is None:
|
|
return {"odpoved": None, "timeout": True}
|
|
return {"odpoved": odp}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
log("MCP Telegram server spuštěn (FastMCP)")
|
|
mcp.run()
|