from __future__ import annotations import json import os import sys from pathlib import Path from typing import Any from app.gitea_client import GiteaClient 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. Use the Gitea tools when the user asks to inspect " "or manage repositories, commits, issues, pull requests, and code-review workflow " "metadata in the user's configured Gitea instance. 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) if name == "gitea_list_repositories": return self.gitea_list_repositories(arguments) if name == "gitea_get_repository": return self.gitea_get_repository(arguments) if name == "gitea_list_issues": return self.gitea_list_issues(arguments) if name == "gitea_create_issue": return self.gitea_create_issue(arguments) if name == "gitea_comment_issue": return self.gitea_comment_issue(arguments) if name == "gitea_list_pull_requests": return self.gitea_list_pull_requests(arguments) if name == "gitea_create_pull_request": return self.gitea_create_pull_request(arguments) if name == "gitea_list_branches": return self.gitea_list_branches(arguments) if name == "gitea_list_commits": return self.gitea_list_commits(arguments) if name == "gitea_get_commit": return self.gitea_get_commit(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 gitea_list_repositories(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() repos = client.list_repositories( limit=int(arguments.get("limit") or 30), page=int(arguments.get("page") or 1), ) if not repos: text = "No Gitea repositories found." else: text = "\n".join(_format_gitea_repo(repo) for repo in repos) return { "content": [{"type": "text", "text": text}], "structuredContent": {"repositories": repos}, } def gitea_get_repository(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) repository = client.get_repository(owner, repo) return { "content": [{"type": "text", "text": _format_gitea_repo(repository)}], "structuredContent": {"repository": repository}, } def gitea_list_issues(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) issues = client.list_issues( owner, repo, state=str(arguments.get("state") or "open"), limit=int(arguments.get("limit") or 20), page=int(arguments.get("page") or 1), ) if not issues: text = "No Gitea issues found." else: text = "\n".join(_format_gitea_issue(issue) for issue in issues) return { "content": [{"type": "text", "text": text}], "structuredContent": {"issues": issues}, } def gitea_create_issue(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) issue = client.create_issue( owner, repo, title=str(arguments["title"]), body=str(arguments.get("body") or ""), ) return { "content": [{"type": "text", "text": f"Created Gitea issue: {_format_gitea_issue(issue)}"}], "structuredContent": {"issue": issue}, } def gitea_comment_issue(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) comment = client.comment_issue( owner, repo, issue_index=int(arguments["issue_index"]), body=str(arguments["body"]), ) url = comment.get("html_url") or comment.get("url") or "" text = f"Added Gitea issue comment: {url}" if url else "Added Gitea issue comment." return { "content": [{"type": "text", "text": text}], "structuredContent": {"comment": comment}, } def gitea_list_pull_requests(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) pulls = client.list_pull_requests( owner, repo, state=str(arguments.get("state") or "open"), limit=int(arguments.get("limit") or 20), page=int(arguments.get("page") or 1), ) if not pulls: text = "No Gitea pull requests found." else: text = "\n".join(_format_gitea_pull_request(pull) for pull in pulls) return { "content": [{"type": "text", "text": text}], "structuredContent": {"pull_requests": pulls}, } def gitea_create_pull_request(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) pull = client.create_pull_request( owner, repo, head=str(arguments["head"]), base=str(arguments.get("base") or "main"), title=str(arguments["title"]), body=str(arguments.get("body") or ""), ) return { "content": [{"type": "text", "text": f"Created Gitea pull request: {_format_gitea_pull_request(pull)}"}], "structuredContent": {"pull_request": pull}, } def gitea_list_branches(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) branches = client.list_branches( owner, repo, limit=int(arguments.get("limit") or 30), page=int(arguments.get("page") or 1), ) if not branches: text = "No Gitea branches found." else: text = "\n".join(str(branch.get("name") or "(unknown)") for branch in branches) return { "content": [{"type": "text", "text": text}], "structuredContent": {"branches": branches}, } def gitea_list_commits(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) commits = client.list_commits( owner, repo, sha=_optional_string(arguments.get("sha") or arguments.get("branch")), path=_optional_string(arguments.get("path")), limit=int(arguments.get("limit") or 20), page=int(arguments.get("page") or 1), ) if not commits: text = "No Gitea commits found." else: text = "\n".join(_format_gitea_commit(commit) for commit in commits) return { "content": [{"type": "text", "text": text}], "structuredContent": {"commits": commits}, } def gitea_get_commit(self, arguments: dict[str, Any]) -> dict[str, Any]: client = GiteaClient.from_env() owner, repo = _repo_arguments(arguments, client) commit = client.get_commit(owner, repo, sha=str(arguments["sha"])) return { "content": [{"type": "text", "text": _format_gitea_commit(commit, include_files=True)}], "structuredContent": {"commit": commit}, } 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": "Telegram chat id safety check. Use the chat_id from bridge context.", }, }, "required": ["schedule_id", "chat_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", "chat_id"], }, "annotations": { "title": "Delete personal record", "readOnlyHint": False, "destructiveHint": True, "idempotentHint": False, "openWorldHint": False, }, }, ] + _gitea_tools() def _gitea_tools() -> list[dict[str, Any]]: repo_properties = { "owner": { "type": "string", "description": "Repository owner. Defaults to GITEA_DEFAULT_OWNER when omitted.", }, "repo": { "type": "string", "description": "Repository name. Defaults to GITEA_DEFAULT_REPO when omitted.", }, } readonly_annotations = { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": False, } write_annotations = { "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False, } return [ { "name": "gitea_list_repositories", "description": "List repositories visible to the configured Gitea token.", "inputSchema": { "type": "object", "properties": { "limit": {"type": "integer", "minimum": 1, "maximum": 100}, "page": {"type": "integer", "minimum": 1}, }, }, "annotations": {"title": "List Gitea repositories", **readonly_annotations}, }, { "name": "gitea_get_repository", "description": "Get metadata for one Gitea repository.", "inputSchema": { "type": "object", "properties": repo_properties, }, "annotations": {"title": "Get Gitea repository", **readonly_annotations}, }, { "name": "gitea_list_issues", "description": "List issues for a Gitea repository.", "inputSchema": { "type": "object", "properties": { **repo_properties, "state": {"type": "string", "enum": ["open", "closed", "all"]}, "limit": {"type": "integer", "minimum": 1, "maximum": 100}, "page": {"type": "integer", "minimum": 1}, }, }, "annotations": {"title": "List Gitea issues", **readonly_annotations}, }, { "name": "gitea_create_issue", "description": "Create an issue in a Gitea repository.", "inputSchema": { "type": "object", "properties": { **repo_properties, "title": {"type": "string"}, "body": {"type": "string"}, }, "required": ["title"], }, "annotations": {"title": "Create Gitea issue", **write_annotations}, }, { "name": "gitea_comment_issue", "description": "Add a comment to a Gitea issue or pull request conversation.", "inputSchema": { "type": "object", "properties": { **repo_properties, "issue_index": {"type": "integer", "minimum": 1}, "body": {"type": "string"}, }, "required": ["issue_index", "body"], }, "annotations": {"title": "Comment on Gitea issue", **write_annotations}, }, { "name": "gitea_list_pull_requests", "description": "List pull requests for a Gitea repository.", "inputSchema": { "type": "object", "properties": { **repo_properties, "state": {"type": "string", "enum": ["open", "closed", "all"]}, "limit": {"type": "integer", "minimum": 1, "maximum": 100}, "page": {"type": "integer", "minimum": 1}, }, }, "annotations": {"title": "List Gitea pull requests", **readonly_annotations}, }, { "name": "gitea_create_pull_request", "description": "Create a pull request in a Gitea repository after a branch has been pushed.", "inputSchema": { "type": "object", "properties": { **repo_properties, "head": { "type": "string", "description": "Source branch, for example codex/my-change.", }, "base": { "type": "string", "description": "Target branch. Defaults to main.", }, "title": {"type": "string"}, "body": {"type": "string"}, }, "required": ["head", "title"], }, "annotations": {"title": "Create Gitea pull request", **write_annotations}, }, { "name": "gitea_list_branches", "description": "List branches for a Gitea repository.", "inputSchema": { "type": "object", "properties": { **repo_properties, "limit": {"type": "integer", "minimum": 1, "maximum": 100}, "page": {"type": "integer", "minimum": 1}, }, }, "annotations": {"title": "List Gitea branches", **readonly_annotations}, }, { "name": "gitea_list_commits", "description": "List recent commits for a Gitea repository, optionally scoped to a branch, SHA, or path.", "inputSchema": { "type": "object", "properties": { **repo_properties, "sha": { "type": "string", "description": "Branch name, tag, or commit SHA to list from. Defaults to the repository default branch.", }, "branch": { "type": "string", "description": "Alias for sha when the user names a branch.", }, "path": { "type": "string", "description": "Optional repository path to filter commits by.", }, "limit": {"type": "integer", "minimum": 1, "maximum": 100}, "page": {"type": "integer", "minimum": 1}, }, }, "annotations": {"title": "List Gitea commits", **readonly_annotations}, }, { "name": "gitea_get_commit", "description": "Get one Gitea commit by SHA, including message, author, stats, and changed files when available.", "inputSchema": { "type": "object", "properties": { **repo_properties, "sha": {"type": "string", "description": "Commit SHA to inspect."}, }, "required": ["sha"], }, "annotations": {"title": "Get Gitea commit", **readonly_annotations}, }, ] def _record_categories() -> list[str]: return ["work_log", "todo", "purchase", "birthday", "contact", "note", "preference"] def _repo_arguments(arguments: dict[str, Any], client: GiteaClient) -> tuple[str, str]: owner = str(arguments.get("owner") or client.config.default_owner or "").strip() repo = str(arguments.get("repo") or client.config.default_repo or "").strip() if not owner: raise ValueError("owner is required or GITEA_DEFAULT_OWNER must be set") if not repo: raise ValueError("repo is required or GITEA_DEFAULT_REPO must be set") return owner, repo def _optional_string(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None def _format_gitea_repo(repo: dict[str, Any]) -> str: full_name = repo.get("full_name") or repo.get("name") or "(unknown)" default_branch = repo.get("default_branch") or repo.get("default_branch_name") or "" clone_url = repo.get("clone_url") or "" html_url = repo.get("html_url") or "" bits = [str(full_name)] if default_branch: bits.append(f"default={default_branch}") if html_url: bits.append(str(html_url)) if clone_url: bits.append(f"clone={clone_url}") return " - ".join(bits) def _format_gitea_issue(issue: dict[str, Any]) -> str: index = issue.get("number") or issue.get("index") or "?" title = issue.get("title") or "(untitled)" state = issue.get("state") or "unknown" html_url = issue.get("html_url") or "" suffix = f" - {html_url}" if html_url else "" return f"#{index} [{state}] {title}{suffix}" def _format_gitea_pull_request(pull: dict[str, Any]) -> str: index = pull.get("number") or pull.get("index") or "?" title = pull.get("title") or "(untitled)" state = pull.get("state") or "unknown" html_url = pull.get("html_url") or "" suffix = f" - {html_url}" if html_url else "" return f"#{index} [{state}] {title}{suffix}" def _format_gitea_commit(commit: dict[str, Any], include_files: bool = False) -> str: sha = str(commit.get("sha") or "")[:10] or "(unknown)" payload = commit.get("commit") if isinstance(commit.get("commit"), dict) else {} message = str(payload.get("message") or commit.get("message") or "(no message)").strip() subject = message.splitlines()[0] if message else "(no message)" author = payload.get("author") if isinstance(payload.get("author"), dict) else {} author_name = author.get("name") or commit.get("author_name") or "" authored_at = commit.get("created") or author.get("date") or "" html_url = commit.get("html_url") or "" stats = commit.get("stats") if isinstance(commit.get("stats"), dict) else {} bits = [sha] if authored_at: bits.append(str(authored_at)) if author_name: bits.append(f"by {author_name}") bits.append(subject) if stats: additions = stats.get("additions") deletions = stats.get("deletions") total = stats.get("total") stat_bits = [] if total is not None: stat_bits.append(f"{total} changed") if additions is not None: stat_bits.append(f"+{additions}") if deletions is not None: stat_bits.append(f"-{deletions}") if stat_bits: bits.append(", ".join(stat_bits)) if html_url: bits.append(str(html_url)) text = " - ".join(bits) if not include_files: return text files = commit.get("files") if not isinstance(files, list) or not files: return text file_lines = [] for changed_file in files[:20]: if not isinstance(changed_file, dict): continue filename = changed_file.get("filename") or changed_file.get("name") status = changed_file.get("status") if filename: file_lines.append(f"- {filename}" + (f" ({status})" if status else "")) if not file_lines: return text if len(files) > len(file_lines): file_lines.append(f"- ... {len(files) - len(file_lines)} more files") return text + "\nChanged files:\n" + "\n".join(file_lines) 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()