Wickedness/nas_connection
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Gitea commit lookup tools

Browse Source
This commit is contained in:
y2keui
2026-06-10 11:07:27 +09:00
parent 75e85887b5
commit 060a6950a4
4 changed files with 172 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@@ -112,7 +112,7 @@ chat id.
## Gitea integration
Set these optional values in `.env` to let Codex manage Gitea repositories,
issues, pull requests, and HTTPS git authentication from Telegram:
commits, issues, pull requests, and HTTPS git authentication from Telegram:
```env
GITEA_BASE_URL=https://gitea.example.com
@@ -124,8 +124,8 @@ GITEA_DEFAULT_REPO=your-default-repo
GITEA_GIT_AUTH_ENABLED=true
```
Codex receives Gitea MCP tools for repository, issue, branch, and pull-request
metadata. The container also configures `GIT_ASKPASS` so HTTPS `git clone`,
Codex receives Gitea MCP tools for repository, commit, issue, branch, and
pull-request metadata. The container also configures `GIT_ASKPASS` so HTTPS `git clone`,
`fetch`, and `push` can use the token without embedding it in command lines.
Example Telegram workflow:

8
app/bot.py
View File

@@ -1331,7 +1331,7 @@ class TelegramCodexBot:
f"{runtime_settings}\n"
"Configured Gitea integration:\n"
f"{gitea_context}\n"
"If the user asks for repository, issue, branch, or pull-request metadata in Gitea, "
"If the user asks for repository, commit, issue, branch, or pull-request metadata in Gitea, "
"use the gitea_* MCP tools. Never reveal GITEA_TOKEN or print commands that embed it. "
"For HTTPS git clone/fetch/push, the container may provide GIT_ASKPASS from the "
"configured Gitea environment.\n"
@@ -1788,7 +1788,11 @@ def _scheduled_prompt(schedule: dict) -> str:
return (
"This is a scheduled Telegram bridge task. Run the task now and produce a concise "
"Telegram-ready response. Do not create another schedule unless the task explicitly "
"asks you to change scheduling.\n\n"
"asks you to change scheduling. If repository history is relevant, prefer the Gitea "
"commit tools. Do not describe a missing tool or unavailable detail as a permission "
"problem unless an API error explicitly says permission was denied. Only include a "
"commit summary when the task asks for repository history or a repository can be "
"identified from the task/context.\n\n"
f"Schedule: {format_schedule(schedule)}\n\n"
f"Task:\n{schedule.get('prompt') or ''}"
)

136
app/bridge_mcp.py
View File

@@ -60,7 +60,7 @@ class BridgeMcpServer:
"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 "
"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 "
@@ -125,6 +125,10 @@ class BridgeMcpServer:
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]:
@@ -396,6 +400,35 @@ class BridgeMcpServer:
"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 [
@@ -737,6 +770,44 @@ def _gitea_tools() -> list[dict[str, Any]]:
},
"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},
},
]
@@ -754,6 +825,13 @@ def _repo_arguments(arguments: dict[str, Any], client: GiteaClient) -> tuple[str
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 ""
@@ -787,6 +865,62 @@ def _format_gitea_pull_request(pull: dict[str, Any]) -> str:
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}

28
app/gitea_client.py
View File

@@ -142,6 +142,34 @@ class GiteaClient:
)
return result if isinstance(result, list) else []
def list_commits(
self,
owner: str,
repo: str,
sha: str | None = None,
path: str | None = None,
limit: int = 20,
page: int = 1,
) -> list[dict[str, Any]]:
query = {"limit": str(limit), "page": str(page)}
if sha:
query["sha"] = sha
if path:
query["path"] = path
result = self._request(
"GET",
f"/repos/{_quote(owner)}/{_quote(repo)}/commits",
query=query,
)
return result if isinstance(result, list) else []
def get_commit(self, owner: str, repo: str, sha: str) -> dict[str, Any]:
result = self._request(
"GET",
f"/repos/{_quote(owner)}/{_quote(repo)}/git/commits/{_quote(sha)}",
)
return result if isinstance(result, dict) else {}
def _request(
self,
method: str,