"""
Tool: gitea_api
Gitea issue helper for a configurable repository (default smk/stargazer-v3).
"""
from __future__ import annotations
import jsonutil as json
import os
from typing import TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
TOOL_NAME = "gitea_api"
TOOL_DESCRIPTION = (
"Gitea API: list, open, comment, or close issues. Repository defaults to "
"smk/stargazer-v3; override with owner/repo arguments or GITEA_DEFAULT_OWNER / "
"GITEA_DEFAULT_REPO. Uses Redis-stored gitea key, GITEA_TOKEN, or GITEA_TOKEN_FILE."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"action": {"type": "string", "enum": ["list", "open", "comment", "close"]},
"issue_id": {"type": "integer"},
"title": {"type": "string"},
"body": {"type": "string"},
"owner": {"type": "string", "description": "Repository owner (optional)"},
"repo": {"type": "string", "description": "Repository name (optional)"},
},
"required": ["action"],
}
[docs]
async def run(
action: str,
issue_id: int | None = None,
title: str | None = None,
body: str | None = None,
owner: str | None = None,
repo: str | None = None,
ctx: ToolContext | None = None,
) -> str:
"""List, open, comment on, or close Gitea issues for a repository.
Args:
action: One of ``list``, ``open``, ``comment``, or ``close``.
issue_id: Issue number (required for ``comment`` and ``close``).
title: Issue title (used by ``open``).
body: Issue or comment body (used by ``open`` and ``comment``).
owner: Repository owner; defaults to ``GITEA_DEFAULT_OWNER`` or ``smk``.
repo: Repository name; defaults to ``GITEA_DEFAULT_REPO`` or ``stargazer-v3``.
ctx: Tool execution context (supplies user, Redis, and config).
Returns:
str: JSON response from the Gitea API, or a JSON error string.
"""
from tools.manage_api_keys import get_gitea_credentials, missing_api_key_error
creds = await get_gitea_credentials(
getattr(ctx, "user_id", "") if ctx else "",
redis_client=getattr(ctx, "redis", None) if ctx else None,
channel_id=getattr(ctx, "channel_id", None) if ctx else None,
fallback_to_pool=True,
config=getattr(ctx, "config", None) if ctx else None,
)
if not creds:
return missing_api_key_error("gitea")
token, base_url = creds
own = (owner or os.environ.get("GITEA_DEFAULT_OWNER") or "smk").strip()
rep = (repo or os.environ.get("GITEA_DEFAULT_REPO") or "stargazer-v3").strip()
root = f"{base_url.rstrip('/')}/api/v1/repos/{own}/{rep}/issues"
headers = {
"Authorization": f"token {token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
async with aiohttp.ClientSession() as session:
if action == "list":
async with session.get(root, headers=headers) as resp:
raw = await resp.text()
if resp.status >= 400:
return raw[:4000]
try:
return json.dumps(json.loads(raw))
except json.JSONDecodeError:
return raw[:4000]
if action == "open":
async with session.post(
root, headers=headers, json={"title": title or "", "body": body or ""}
) as resp:
raw = await resp.text()
try:
return json.dumps(json.loads(raw))
except json.JSONDecodeError:
return raw[:4000]
if action == "comment":
if issue_id is None:
return json.dumps({"error": "issue_id required for comment"})
url = f"{root}/{issue_id}/comments"
async with session.post(
url, headers=headers, json={"body": body or ""}
) as resp:
raw = await resp.text()
try:
return json.dumps(json.loads(raw))
except json.JSONDecodeError:
return raw[:4000]
if action == "close":
if issue_id is None:
return json.dumps({"error": "issue_id required for close"})
url = f"{root}/{issue_id}"
async with session.patch(
url, headers=headers, json={"state": "closed"}
) as resp:
raw = await resp.text()
try:
return json.dumps(json.loads(raw))
except json.JSONDecodeError:
return raw[:4000]
return json.dumps({"error": f"unknown action: {action}"})