804 lines
32 KiB
Python
804 lines
32 KiB
Python
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, 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)
|
|
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 _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},
|
|
},
|
|
]
|
|
|
|
|
|
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 _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 _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()
|