from __future__ import annotations import json import os import sys from pathlib import Path from typing import Any from app.state import StateStore, format_personal_record, format_schedule PROTOCOL_VERSION = "2024-11-05" def main() -> None: server = BridgeMcpServer( StateStore(Path(os.getenv("BOT_STATE_PATH", "/app/data/state.json"))), default_timezone=os.getenv("BOT_TIMEZONE", "Asia/Seoul"), ) server.run() class BridgeMcpServer: def __init__(self, state: StateStore, default_timezone: str) -> None: self.state = state self.default_timezone = default_timezone def run(self) -> None: for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) response = self.handle(message) except Exception as exc: response = _error(None, -32603, str(exc)) if response is not None: sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n") sys.stdout.flush() def handle(self, message: dict[str, Any]) -> dict[str, Any] | None: method = message.get("method") request_id = message.get("id") if method == "initialize": params = message.get("params") or {} return _result( request_id, { "protocolVersion": params.get("protocolVersion") or PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": { "name": "telegram-codex-bridge", "version": "0.1.0", }, "instructions": ( "Use these tools when the user asks to schedule recurring Telegram " "notifications, manage bridge schedules, or persist personal assistant " "records such as work history, todos, purchase candidates, birthdays, " "contacts, and notes. The Telegram chat_id and " "current thread_id are provided in the bridge context inside each user turn. " "Use that chat_id as the default notification target and do not ask for a " "destination unless the user explicitly wants a different chat." ), }, ) if method == "notifications/initialized": return None if method == "tools/list": return _result(request_id, {"tools": _tools()}) if method == "tools/call": params = message.get("params") or {} name = params.get("name") arguments = params.get("arguments") or {} try: return _result(request_id, self.call_tool(str(name), arguments)) except Exception as exc: return _result( request_id, { "content": [{"type": "text", "text": f"Tool failed: {exc}"}], "isError": True, }, ) if request_id is None: return None return _error(request_id, -32601, f"Unknown method: {method}") def call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: if name == "telegram_create_schedule": return self.create_schedule(arguments) if name == "telegram_list_schedules": return self.list_schedules(arguments) if name == "telegram_delete_schedule": return self.delete_schedule(arguments) if name == "telegram_save_personal_record": return self.save_personal_record(arguments) if name == "telegram_list_personal_records": return self.list_personal_records(arguments) if name == "telegram_update_personal_record": return self.update_personal_record(arguments) if name == "telegram_delete_personal_record": return self.delete_personal_record(arguments) raise ValueError(f"Unknown tool: {name}") def create_schedule(self, arguments: dict[str, Any]) -> dict[str, Any]: chat_id = int(arguments["chat_id"]) schedule_type = str(arguments.get("schedule_type") or "daily") if schedule_type not in {"daily", "interval"}: raise ValueError("schedule_type must be daily or interval") schedule: dict[str, Any] = { "chat_id": chat_id, "thread_id": arguments.get("thread_id") or None, "name": str(arguments.get("name") or "Scheduled Codex task"), "prompt": str(arguments["prompt"]), "schedule_type": schedule_type, "timezone": str(arguments.get("timezone") or self.default_timezone), "status": "active", } if schedule_type == "interval": schedule["interval_minutes"] = max(1, int(arguments.get("interval_minutes") or 60)) else: schedule["hour"] = int(arguments.get("hour") if arguments.get("hour") is not None else 8) schedule["minute"] = int(arguments.get("minute") if arguments.get("minute") is not None else 0) created = self.state.add_schedule(schedule) text = f"Created Telegram schedule: {format_schedule(created)}" return { "content": [{"type": "text", "text": text}], "structuredContent": {"schedule": created}, } def list_schedules(self, arguments: dict[str, Any]) -> dict[str, Any]: chat_id = arguments.get("chat_id") schedules = self.state.list_schedules( chat_id=int(chat_id) if chat_id is not None else None, include_paused=bool(arguments.get("include_paused", True)), ) if not schedules: text = "No Telegram bridge schedules are registered." else: text = "\n".join(format_schedule(schedule) for schedule in schedules) return { "content": [{"type": "text", "text": text}], "structuredContent": {"schedules": schedules}, } def delete_schedule(self, arguments: dict[str, Any]) -> dict[str, Any]: schedule_id = int(arguments["schedule_id"]) chat_id = arguments.get("chat_id") removed = self.state.remove_schedule( schedule_id, chat_id=int(chat_id) if chat_id is not None else None, ) if removed is None: raise ValueError(f"schedule not found: {schedule_id}") text = f"Deleted Telegram schedule: {format_schedule(removed)}" return { "content": [{"type": "text", "text": text}], "structuredContent": {"deleted": removed}, } def save_personal_record(self, arguments: dict[str, Any]) -> dict[str, Any]: chat_id = int(arguments["chat_id"]) category = str(arguments.get("category") or "note") if category not in _record_categories(): raise ValueError(f"category must be one of: {', '.join(_record_categories())}") tags = arguments.get("tags") or [] if not isinstance(tags, list): tags = [str(tags)] metadata = arguments.get("metadata") or {} if not isinstance(metadata, dict): metadata = {} record = { "chat_id": chat_id, "category": category, "title": str(arguments.get("title") or "Untitled"), "content": str(arguments.get("content") or ""), "status": str(arguments.get("status") or "active"), "date": arguments.get("date") or None, "start_date": arguments.get("start_date") or None, "end_date": arguments.get("end_date") or None, "due_date": arguments.get("due_date") or None, "tags": [str(tag) for tag in tags], "metadata": metadata, } created = self.state.add_personal_record(record) text = f"Saved personal record: {format_personal_record(created)}" return { "content": [{"type": "text", "text": text}], "structuredContent": {"record": created}, } def list_personal_records(self, arguments: dict[str, Any]) -> dict[str, Any]: chat_id = arguments.get("chat_id") records = self.state.list_personal_records( chat_id=int(chat_id) if chat_id is not None else None, category=arguments.get("category") or None, status=arguments.get("status") or None, query=arguments.get("query") or None, limit=int(arguments.get("limit") or 30), ) if not records: text = "No personal records found." else: text = "\n".join(format_personal_record(record) for record in records) return { "content": [{"type": "text", "text": text}], "structuredContent": {"records": records}, } def update_personal_record(self, arguments: dict[str, Any]) -> dict[str, Any]: record_id = int(arguments["record_id"]) chat_id = arguments.get("chat_id") updates = dict(arguments.get("updates") or {}) if "category" in updates and updates["category"] not in _record_categories(): raise ValueError(f"category must be one of: {', '.join(_record_categories())}") if "tags" in updates and not isinstance(updates["tags"], list): updates["tags"] = [str(updates["tags"])] updated = self.state.update_personal_record( record_id, updates, chat_id=int(chat_id) if chat_id is not None else None, ) if updated is None: raise ValueError(f"personal record not found: {record_id}") text = f"Updated personal record: {format_personal_record(updated)}" return { "content": [{"type": "text", "text": text}], "structuredContent": {"record": updated}, } def delete_personal_record(self, arguments: dict[str, Any]) -> dict[str, Any]: record_id = int(arguments["record_id"]) chat_id = arguments.get("chat_id") removed = self.state.remove_personal_record( record_id, chat_id=int(chat_id) if chat_id is not None else None, ) if removed is None: raise ValueError(f"personal record not found: {record_id}") text = f"Deleted personal record: {format_personal_record(removed)}" return { "content": [{"type": "text", "text": text}], "structuredContent": {"deleted": removed}, } def _tools() -> list[dict[str, Any]]: return [ { "name": "telegram_create_schedule", "description": ( "Create a recurring Telegram notification schedule. Use this when the user asks " "Codex to remind, notify, monitor, or run a recurring task through Telegram. " "The prompt should describe only the work to run at schedule time, not the cadence." ), "inputSchema": { "type": "object", "properties": { "chat_id": { "type": "integer", "description": "Telegram chat id that should receive the scheduled result.", }, "thread_id": { "type": "string", "description": "Current Codex app-server thread id, if available.", }, "name": {"type": "string", "description": "Short schedule name."}, "prompt": { "type": "string", "description": "Task prompt to run at the scheduled time.", }, "schedule_type": { "type": "string", "enum": ["daily", "interval"], "description": "daily for wall-clock time, interval for every N minutes.", }, "hour": { "type": "integer", "minimum": 0, "maximum": 23, "description": "Daily run hour in the selected timezone.", }, "minute": { "type": "integer", "minimum": 0, "maximum": 59, "description": "Daily run minute in the selected timezone.", }, "interval_minutes": { "type": "integer", "minimum": 1, "description": "Interval cadence in minutes.", }, "timezone": { "type": "string", "description": "IANA timezone, usually Asia/Seoul.", }, }, "required": ["chat_id", "name", "prompt", "schedule_type"], }, "annotations": { "title": "Create Telegram schedule", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False, }, }, { "name": "telegram_list_schedules", "description": "List Telegram bridge schedules, optionally scoped to a chat_id.", "inputSchema": { "type": "object", "properties": { "chat_id": {"type": "integer"}, "include_paused": {"type": "boolean"}, }, }, "annotations": { "title": "List Telegram schedules", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False, }, }, { "name": "telegram_delete_schedule", "description": "Delete a Telegram bridge schedule by id.", "inputSchema": { "type": "object", "properties": { "schedule_id": {"type": "integer"}, "chat_id": { "type": "integer", "description": "Optional Telegram chat id safety check.", }, }, "required": ["schedule_id"], }, "annotations": { "title": "Delete Telegram schedule", "readOnlyHint": False, "destructiveHint": True, "idempotentHint": False, "openWorldHint": False, }, }, { "name": "telegram_save_personal_record", "description": ( "Save a durable personal assistant record to NAS-backed Telegram bridge storage. " "Use for work history, todos, purchase plans, birthdays, contacts, personal notes, " "preferences, and other facts the user explicitly wants remembered later." ), "inputSchema": { "type": "object", "properties": { "chat_id": {"type": "integer"}, "category": { "type": "string", "enum": _record_categories(), "description": "Record category.", }, "title": {"type": "string", "description": "Short title."}, "content": {"type": "string", "description": "Full details to remember."}, "status": { "type": "string", "description": "active, done, cancelled, archived, or another short status.", }, "date": {"type": "string", "description": "Primary date, ISO YYYY-MM-DD if possible."}, "start_date": {"type": "string", "description": "Start date, ISO YYYY-MM-DD if possible."}, "end_date": {"type": "string", "description": "End date, ISO YYYY-MM-DD if possible."}, "due_date": {"type": "string", "description": "Due date, ISO YYYY-MM-DD if possible."}, "tags": {"type": "array", "items": {"type": "string"}}, "metadata": {"type": "object", "additionalProperties": True}, }, "required": ["chat_id", "category", "title", "content"], }, "annotations": { "title": "Save personal record", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False, }, }, { "name": "telegram_list_personal_records", "description": "List or search personal assistant records stored by the Telegram bridge.", "inputSchema": { "type": "object", "properties": { "chat_id": {"type": "integer"}, "category": {"type": "string", "enum": _record_categories()}, "status": {"type": "string"}, "query": {"type": "string"}, "limit": {"type": "integer", "minimum": 1, "maximum": 100}, }, }, "annotations": { "title": "List personal records", "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False, }, }, { "name": "telegram_update_personal_record", "description": "Update a personal assistant record by id.", "inputSchema": { "type": "object", "properties": { "record_id": {"type": "integer"}, "chat_id": {"type": "integer"}, "updates": {"type": "object", "additionalProperties": True}, }, "required": ["record_id", "updates"], }, "annotations": { "title": "Update personal record", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False, }, }, { "name": "telegram_delete_personal_record", "description": "Delete a personal assistant record by id.", "inputSchema": { "type": "object", "properties": { "record_id": {"type": "integer"}, "chat_id": {"type": "integer"}, }, "required": ["record_id"], }, "annotations": { "title": "Delete personal record", "readOnlyHint": False, "destructiveHint": True, "idempotentHint": False, "openWorldHint": False, }, }, ] def _record_categories() -> list[str]: return ["work_log", "todo", "purchase", "birthday", "contact", "note", "preference"] def _result(request_id: Any, result: dict[str, Any]) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": request_id, "result": result} def _error(request_id: Any, code: int, message: str) -> dict[str, Any]: return { "jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}, } if __name__ == "__main__": main()