"""GitHub API tools using per-user OAuth tokens.
Provides repository, issue, PR, gist, notification, and code-search
operations via the GitHub REST API. Requires the user to have connected
their GitHub account via the OAuth flow (connect_service tool).
"""
from __future__ import annotations
import jsonutil as json
import logging
from typing import Any, TYPE_CHECKING
import aiohttp
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
GITHUB_API = "https://api.github.com"
async def _gh_request(
method: str,
path: str,
token: str,
*,
params: dict[str, Any] | None = None,
json_body: dict[str, Any] | None = None,
) -> dict[str, Any] | list[Any] | str:
"""Perform an authenticated GitHub REST API request and parse the response.
Low-level HTTP helper shared by every tool handler in this module. It
attaches the bearer token and the standard GitHub ``Accept`` /
``X-GitHub-Api-Version`` headers, issues the request against
``https://api.github.com``, and normalises the result into a Python object.
A ``204 No Content`` becomes a success marker, any ``>= 400`` status becomes
a structured error dict (with the response body truncated to 2000 chars),
and otherwise the JSON body is decoded (falling back to a 4000-char raw text
snippet if the body is not valid JSON).
Interactions: opens a fresh :class:`aiohttp.ClientSession` per call and sends
one HTTP request to the GitHub API; performs no Redis, event-bus, or other
side effects beyond that network call. Called by every ``github_*`` handler
in this file (e.g. :func:`github_list_repos`, :func:`github_create_issue`,
:func:`github_star_repo`); it has no callers outside this module.
Args:
method: HTTP verb to use (``"GET"``, ``"POST"``, ``"PUT"``,
``"DELETE"``, etc.).
path: API path appended to ``GITHUB_API`` (e.g. ``"/user/repos"``).
token: GitHub OAuth bearer token for the requesting user.
params: Optional query-string parameters.
json_body: Optional JSON request body for write operations.
Returns:
dict[str, Any] | list[Any] | str: The decoded JSON response (dict or
list); ``{"status": "success", "code": 204}`` on ``204``; an
``{"error": ..., "detail": ...}`` dict on HTTP errors; or a truncated
raw string when the body cannot be JSON-decoded.
"""
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
url = f"{GITHUB_API}{path}"
async with aiohttp.ClientSession() as session:
async with session.request(
method, url, headers=headers, params=params, json=json_body
) as resp:
if resp.status == 204:
return {"status": "success", "code": 204}
body = await resp.text()
if resp.status >= 400:
return {
"error": f"GitHub API error ({resp.status})",
"detail": body[:2000],
}
try:
return json.loads(body)
except json.JSONDecodeError:
return body[:4000]
async def _get_token(ctx: ToolContext | None) -> str:
"""Resolve the requesting user's GitHub OAuth access token.
Validates that a usable :class:`ToolContext` with a Redis handle is present,
then delegates to the OAuth manager to fetch (and refresh if needed) a valid
token for the ``"github"`` provider scoped to the current user.
Interactions: calls :func:`oauth_manager.require_oauth_token`, which reads
the stored OAuth credentials from Redis and may raise
:class:`oauth_manager.OAuthNotConnected` containing a connect link when the
user has not linked their GitHub account. Called at the top of every
``github_*`` handler in this module to obtain the token passed to
:func:`_gh_request`.
Args:
ctx: The tool invocation context carrying the Redis client and user
identity; may be ``None`` if the tool was dispatched without one.
Returns:
str: A valid GitHub bearer token for the current user.
Raises:
RuntimeError: If ``ctx`` is ``None`` or has no Redis connection.
oauth_manager.OAuthNotConnected: Propagated from
:func:`require_oauth_token` when the user has no connected GitHub
account (each handler catches this and returns its message).
"""
if ctx is None or ctx.redis is None:
raise RuntimeError("Context or Redis not available")
from oauth_manager import require_oauth_token
return await require_oauth_token(ctx, "github")
def _fmt_repo(r: dict) -> dict:
"""Reduce a raw GitHub repository object to a compact summary dict.
Projects only the fields useful to the model (name, description, visibility,
language, star/fork/issue counts, URL, and last-updated timestamp) so the
tool output stays small. Performs no I/O. Called by :func:`github_list_repos`
to format each repository in the listing.
Args:
r: A repository object as returned by the GitHub REST API.
Returns:
dict: A flattened summary with keys ``full_name``, ``description``,
``private``, ``language``, ``stars``, ``forks``, ``open_issues``,
``url``, and ``updated_at``.
"""
return {
"full_name": r.get("full_name"),
"description": r.get("description"),
"private": r.get("private"),
"language": r.get("language"),
"stars": r.get("stargazers_count"),
"forks": r.get("forks_count"),
"open_issues": r.get("open_issues_count"),
"url": r.get("html_url"),
"updated_at": r.get("updated_at"),
}
def _fmt_issue(i: dict) -> dict:
"""Reduce a raw GitHub issue object to a compact summary dict.
Projects the model-relevant issue fields, flattening the author to a login
string and labels to a list of names, and truncating the body to 1000
characters to keep tool output small. Performs no I/O. Called by
:func:`github_create_issue` (to format the created issue) and
:func:`github_list_issues` (to format each listed issue).
Args:
i: An issue object as returned by the GitHub REST API.
Returns:
dict: A flattened summary with keys ``number``, ``title``, ``state``,
``user``, ``labels``, ``created_at``, ``updated_at``, ``url``, and a
truncated ``body``.
"""
return {
"number": i.get("number"),
"title": i.get("title"),
"state": i.get("state"),
"user": i.get("user", {}).get("login"),
"labels": [l.get("name") for l in i.get("labels", [])],
"created_at": i.get("created_at"),
"updated_at": i.get("updated_at"),
"url": i.get("html_url"),
"body": (i.get("body") or "")[:1000],
}
def _fmt_pr(p: dict) -> dict:
"""Reduce a raw GitHub pull-request object to a compact summary dict.
Projects the model-relevant PR fields, including draft/merged flags and the
diff stats (additions, deletions, changed files) that are only populated on
single-PR responses. Performs no I/O. Called by
:func:`github_list_pull_requests` (to format each listed PR) and
:func:`github_get_pull_request` (to format the single fetched PR).
Args:
p: A pull-request object as returned by the GitHub REST API.
Returns:
dict: A flattened summary with keys ``number``, ``title``, ``state``,
``user``, ``draft``, ``merged``, ``created_at``, ``updated_at``,
``url``, ``additions``, ``deletions``, and ``changed_files``.
"""
return {
"number": p.get("number"),
"title": p.get("title"),
"state": p.get("state"),
"user": p.get("user", {}).get("login"),
"draft": p.get("draft"),
"merged": p.get("merged"),
"created_at": p.get("created_at"),
"updated_at": p.get("updated_at"),
"url": p.get("html_url"),
"additions": p.get("additions"),
"deletions": p.get("deletions"),
"changed_files": p.get("changed_files"),
}
# ---------------------------------------------------------------------------
# Tool handlers
# ---------------------------------------------------------------------------
[docs]
async def github_list_repos(
visibility: str = "all",
sort: str = "updated",
per_page: int = 20,
page: int = 1,
ctx: ToolContext | None = None,
) -> str:
"""List the authenticated user's GitHub repositories as a JSON string.
Tool handler backing the ``github_list_repos`` tool. Fetches the current
user's repositories with optional visibility/sort filters and pagination,
returning a JSON-encoded ``{"count", "repos"}`` payload of compacted
repository summaries.
Interactions: obtains the user's token via :func:`_get_token` (which calls
:func:`oauth_manager.require_oauth_token` against Redis), issues a
``GET /user/repos`` through :func:`_gh_request`, and maps results through
:func:`_fmt_repo`. Registered in the module ``TOOLS`` list and dispatched by
the tool execution layer via ``tool_loader.py``; not called directly by
other Python modules.
Args:
visibility: Repository visibility filter (``"all"``, ``"public"``, or
``"private"``).
sort: Sort order (``"created"``, ``"updated"``, ``"pushed"``, or
``"full_name"``).
per_page: Results per page; clamped to a maximum of 100.
page: 1-based page number.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of ``{"count", "repos"}`` on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {
"visibility": visibility,
"sort": sort,
"per_page": min(per_page, 100),
"page": page,
}
data = await _gh_request("GET", "/user/repos", token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
repos = [_fmt_repo(r) for r in data] if isinstance(data, list) else data
return json.dumps({"count": len(repos), "repos": repos})
[docs]
async def github_search_code(
query: str,
per_page: int = 10,
page: int = 1,
ctx: ToolContext | None = None,
) -> str:
"""Search code across GitHub repositories and return matches as JSON.
Tool handler backing the ``github_search_code`` tool. Runs a GitHub code
search using GitHub's query syntax (supporting qualifiers like ``repo:``,
``language:``, ``path:``) and returns a JSON-encoded ``{"total_count",
"items"}`` payload where each item is reduced to its name, path,
repository, and URL.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup) and issues a ``GET /search/code`` through :func:`_gh_request`.
Registered in the module ``TOOLS`` list and dispatched by the tool execution
layer via ``tool_loader.py``; not called directly by other Python modules.
Args:
query: GitHub code-search query string with optional qualifiers.
per_page: Results per page; clamped to a maximum of 100.
page: 1-based page number.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of ``{"total_count", "items"}`` on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {"q": query, "per_page": min(per_page, 100), "page": page}
data = await _gh_request("GET", "/search/code", token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
items = []
for item in data.get("items", []) if isinstance(data, dict) else []:
items.append(
{
"name": item.get("name"),
"path": item.get("path"),
"repository": item.get("repository", {}).get("full_name"),
"url": item.get("html_url"),
}
)
return json.dumps({"total_count": data.get("total_count", 0), "items": items})
[docs]
async def github_create_issue(
owner: str,
repo: str,
title: str,
body: str = "",
labels: list[str] | None = None,
assignees: list[str] | None = None,
ctx: ToolContext | None = None,
) -> str:
"""Create a new issue on a GitHub repository and return it as JSON.
Tool handler backing the ``github_create_issue`` tool. Builds the issue
payload from the title plus optional body, labels, and assignees, posts it,
and returns the created issue in compacted form.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup), issues a ``POST /repos/{owner}/{repo}/issues`` through
:func:`_gh_request`, and formats the result with :func:`_fmt_issue`.
Registered in the module ``TOOLS`` list and dispatched by the tool execution
layer via ``tool_loader.py``; not called directly by other Python modules.
Args:
owner: Repository owner (user or organization).
repo: Repository name.
title: Issue title.
body: Optional issue body in Markdown; omitted from the payload if empty.
labels: Optional list of label names to apply.
assignees: Optional list of usernames to assign.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of the created issue summary on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
payload: dict[str, Any] = {"title": title}
if body:
payload["body"] = body
if labels:
payload["labels"] = labels
if assignees:
payload["assignees"] = assignees
data = await _gh_request(
"POST", f"/repos/{owner}/{repo}/issues", token, json_body=payload
)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps(_fmt_issue(data) if isinstance(data, dict) else data)
[docs]
async def github_list_issues(
owner: str,
repo: str,
state: str = "open",
labels: str = "",
per_page: int = 20,
page: int = 1,
ctx: ToolContext | None = None,
) -> str:
"""List issues on a GitHub repository as a JSON string.
Tool handler backing the ``github_list_issues`` tool. Fetches issues
filtered by state and (optionally) a comma-separated label list, with
pagination, returning a JSON-encoded ``{"count", "issues"}`` payload of
compacted issue summaries. Note that the GitHub issues endpoint also
includes pull requests.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup), issues a ``GET /repos/{owner}/{repo}/issues`` through
:func:`_gh_request`, and maps results through :func:`_fmt_issue`. Registered
in the module ``TOOLS`` list and dispatched by the tool execution layer via
``tool_loader.py``; not called directly by other Python modules.
Args:
owner: Repository owner.
repo: Repository name.
state: Issue state filter (``"open"``, ``"closed"``, or ``"all"``).
labels: Optional comma-separated label names to filter by; omitted when
empty.
per_page: Results per page; clamped to a maximum of 100.
page: 1-based page number.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of ``{"count", "issues"}`` on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params: dict[str, Any] = {
"state": state,
"per_page": min(per_page, 100),
"page": page,
}
if labels:
params["labels"] = labels
data = await _gh_request(
"GET", f"/repos/{owner}/{repo}/issues", token, params=params
)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
issues = [_fmt_issue(i) for i in data] if isinstance(data, list) else data
return json.dumps({"count": len(issues), "issues": issues})
[docs]
async def github_list_pull_requests(
owner: str,
repo: str,
state: str = "open",
sort: str = "created",
per_page: int = 20,
page: int = 1,
ctx: ToolContext | None = None,
) -> str:
"""List pull requests on a GitHub repository as a JSON string.
Tool handler backing the ``github_list_pull_requests`` tool. Fetches PRs
filtered by state and ordered by the requested sort, with pagination,
returning a JSON-encoded ``{"count", "pull_requests"}`` payload of compacted
PR summaries.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup), issues a ``GET /repos/{owner}/{repo}/pulls`` through
:func:`_gh_request`, and maps results through :func:`_fmt_pr`. Registered in
the module ``TOOLS`` list and dispatched by the tool execution layer via
``tool_loader.py``; not called directly by other Python modules.
Args:
owner: Repository owner.
repo: Repository name.
state: PR state filter (``"open"``, ``"closed"``, or ``"all"``).
sort: Sort order (``"created"``, ``"updated"``, ``"popularity"``, or
``"long-running"``).
per_page: Results per page; clamped to a maximum of 100.
page: 1-based page number.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of ``{"count", "pull_requests"}`` on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {
"state": state,
"sort": sort,
"per_page": min(per_page, 100),
"page": page,
}
data = await _gh_request(
"GET", f"/repos/{owner}/{repo}/pulls", token, params=params
)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
prs = [_fmt_pr(p) for p in data] if isinstance(data, list) else data
return json.dumps({"count": len(prs), "pull_requests": prs})
[docs]
async def github_get_pull_request(
owner: str,
repo: str,
pull_number: int,
include_diff: bool = False,
ctx: ToolContext | None = None,
) -> str:
"""Fetch a single pull request (optionally with its diff) as JSON.
Tool handler backing the ``github_get_pull_request`` tool. Returns the
compacted PR summary plus a body excerpt (first 2000 chars), and when
``include_diff`` is set, appends up to the first 8000 characters of the
unified diff fetched with the ``application/vnd.github.v3.diff`` media type.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup), issues a ``GET /repos/{owner}/{repo}/pulls/{pull_number}``
through :func:`_gh_request` and formats it with :func:`_fmt_pr`; for the diff
it opens a separate :class:`aiohttp.ClientSession` and re-requests the same
PR URL with the diff ``Accept`` header. Registered in the module ``TOOLS``
list and dispatched by the tool execution layer via ``tool_loader.py``; not
called directly by other Python modules.
Args:
owner: Repository owner.
repo: Repository name.
pull_number: The pull request number.
include_diff: When ``True``, include a truncated unified diff under the
``"diff"`` key (only added if the diff request returns HTTP 200).
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of the PR summary (with ``body`` and optional
``diff``) on success; the ``OAuthNotConnected`` message when GitHub is
not linked; or a JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
data = await _gh_request("GET", f"/repos/{owner}/{repo}/pulls/{pull_number}", token)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
result = _fmt_pr(data) if isinstance(data, dict) else {"raw": data}
if isinstance(data, dict):
result["body"] = (data.get("body") or "")[:2000]
if include_diff:
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3.diff",
}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{GITHUB_API}/repos/{owner}/{repo}/pulls/{pull_number}",
headers=headers,
) as resp:
if resp.status == 200:
diff = await resp.text()
result["diff"] = diff[:8000]
return json.dumps(result)
[docs]
async def github_create_gist(
description: str,
files: dict[str, str],
public: bool = False,
ctx: ToolContext | None = None,
) -> str:
"""Create a GitHub Gist from one or more named files and return it as JSON.
Tool handler backing the ``github_create_gist`` tool. Wraps each
``filename -> content`` entry into the GitHub gist file structure, creates
the gist, and returns a compact summary of the result.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup) and issues a ``POST /gists`` through :func:`_gh_request`.
Registered in the module ``TOOLS`` list and dispatched by the tool execution
layer via ``tool_loader.py``; not called directly by other Python modules.
Args:
description: Human-readable description of the gist.
files: Mapping of filename to file content; each becomes one gist file.
public: Whether the gist is publicly visible (defaults to ``False``).
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string with the new gist's ``id``, ``url``, ``description``,
``public`` flag, ``files`` (filenames), and ``created_at`` on success;
the ``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
gist_files = {name: {"content": content} for name, content in files.items()}
payload = {"description": description, "public": public, "files": gist_files}
data = await _gh_request("POST", "/gists", token, json_body=payload)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
return json.dumps(
{
"id": data.get("id"),
"url": data.get("html_url"),
"description": data.get("description"),
"public": data.get("public"),
"files": list(data.get("files", {}).keys()),
"created_at": data.get("created_at"),
}
)
[docs]
async def github_get_file(
owner: str,
repo: str,
path: str,
ref: str = "",
ctx: ToolContext | None = None,
) -> str:
"""Fetch a repository file's contents or a directory listing as JSON.
Tool handler backing the ``github_get_file`` tool. For a file, decodes the
base64 payload to UTF-8 (replacing undecodable bytes, falling back to
``"(binary file)"`` on failure) and returns the first 16000 characters of
content with metadata. For a directory, returns a listing of its entries.
An optional ``ref`` pins the lookup to a branch, tag, or commit SHA.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup), issues a ``GET /repos/{owner}/{repo}/contents/{path}`` through
:func:`_gh_request`, and uses the standard-library :mod:`base64` to decode
file content. Registered in the module ``TOOLS`` list and dispatched by the
tool execution layer via ``tool_loader.py``; not called directly by other
Python modules.
Args:
owner: Repository owner.
repo: Repository name.
path: File or directory path within the repository.
ref: Optional branch, tag, or commit SHA; defaults to the repo's
default branch when empty.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of file metadata and truncated content for a file; a
``{"type": "directory", "entries": [...]}`` listing for a directory; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {}
if ref:
params["ref"] = ref
data = await _gh_request(
"GET", f"/repos/{owner}/{repo}/contents/{path}", token, params=params
)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
if isinstance(data, dict) and data.get("type") == "file":
import base64
content_b64 = data.get("content", "")
try:
content = base64.b64decode(content_b64).decode("utf-8", errors="replace")
except Exception:
content = "(binary file)"
return json.dumps(
{
"name": data.get("name"),
"path": data.get("path"),
"size": data.get("size"),
"sha": data.get("sha"),
"content": content[:16000],
"url": data.get("html_url"),
}
)
if isinstance(data, list):
entries = [
{
"name": e.get("name"),
"type": e.get("type"),
"size": e.get("size"),
"path": e.get("path"),
}
for e in data
]
return json.dumps({"type": "directory", "entries": entries})
return json.dumps(data if isinstance(data, dict) else {"raw": str(data)[:4000]})
[docs]
async def github_list_notifications(
all_notifications: bool = False,
per_page: int = 20,
page: int = 1,
ctx: ToolContext | None = None,
) -> str:
"""List the user's GitHub notifications as a JSON string.
Tool handler backing the ``github_list_notifications`` tool. Returns a
JSON-encoded ``{"count", "notifications"}`` payload of compacted
notification threads (id, reason, unread flag, subject title/type,
repository, and timestamp). By default only unread notifications are
returned unless ``all_notifications`` is set.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup) and issues a ``GET /notifications`` through
:func:`_gh_request`. Registered in the module ``TOOLS`` list and dispatched
by the tool execution layer via ``tool_loader.py``; not called directly by
other Python modules.
Args:
all_notifications: When ``True``, include read notifications as well;
sent to the API as the lowercased ``all`` query parameter.
per_page: Results per page; clamped to a maximum of 50.
page: 1-based page number.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string of ``{"count", "notifications"}`` on success; the
``OAuthNotConnected`` message when GitHub is not linked; or a
JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
params = {
"all": str(all_notifications).lower(),
"per_page": min(per_page, 50),
"page": page,
}
data = await _gh_request("GET", "/notifications", token, params=params)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
notifs = []
for n in data if isinstance(data, list) else []:
notifs.append(
{
"id": n.get("id"),
"reason": n.get("reason"),
"unread": n.get("unread"),
"subject_title": n.get("subject", {}).get("title"),
"subject_type": n.get("subject", {}).get("type"),
"repository": n.get("repository", {}).get("full_name"),
"updated_at": n.get("updated_at"),
}
)
return json.dumps({"count": len(notifs), "notifications": notifs})
[docs]
async def github_star_repo(
owner: str,
repo: str,
star: bool = True,
ctx: ToolContext | None = None,
) -> str:
"""Star or unstar a GitHub repository and return the outcome as JSON.
Tool handler backing the ``github_star_repo`` tool. Issues a ``PUT`` to star
or a ``DELETE`` to unstar the given repository, then reports the action
taken.
Interactions: obtains the user's token via :func:`_get_token` (Redis-backed
OAuth lookup) and issues a ``PUT``/``DELETE /user/starred/{owner}/{repo}``
through :func:`_gh_request` (a successful call returns HTTP 204). Registered
in the module ``TOOLS`` list and dispatched by the tool execution layer via
``tool_loader.py``; not called directly by other Python modules.
Args:
owner: Repository owner.
repo: Repository name.
star: ``True`` to star the repo, ``False`` to unstar it.
ctx: The tool invocation context providing Redis and user identity.
Returns:
str: A JSON string ``{"status": "success", "action", "repo"}`` on
success; the ``OAuthNotConnected`` message when GitHub is not linked; or
a JSON-encoded error dict on API failure.
"""
from oauth_manager import OAuthNotConnected
try:
token = await _get_token(ctx)
except OAuthNotConnected as e:
return str(e)
method = "PUT" if star else "DELETE"
data = await _gh_request(method, f"/user/starred/{owner}/{repo}", token)
if isinstance(data, dict) and "error" in data:
return json.dumps(data)
action = "starred" if star else "unstarred"
return json.dumps(
{"status": "success", "action": action, "repo": f"{owner}/{repo}"}
)
# ---------------------------------------------------------------------------
# Tool registration (multi-tool format)
# ---------------------------------------------------------------------------
TOOLS = [
{
"name": "github_list_repos",
"description": "List the authenticated user's GitHub repositories with optional filters for visibility and sort order.",
"parameters": {
"type": "object",
"properties": {
"visibility": {
"type": "string",
"enum": ["all", "public", "private"],
"description": "Filter by visibility",
},
"sort": {
"type": "string",
"enum": ["created", "updated", "pushed", "full_name"],
"description": "Sort order",
},
"per_page": {
"type": "integer",
"description": "Results per page (max 100)",
},
"page": {"type": "integer", "description": "Page number"},
},
},
"handler": github_list_repos,
},
{
"name": "github_search_code",
"description": "Search for code across GitHub repositories. Uses GitHub code search syntax (e.g. 'addClass repo:jquery/jquery').",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "GitHub search query (supports qualifiers like repo:, language:, path:)",
},
"per_page": {
"type": "integer",
"description": "Results per page (max 100)",
},
"page": {"type": "integer", "description": "Page number"},
},
"required": ["query"],
},
"handler": github_search_code,
},
{
"name": "github_create_issue",
"description": "Create a new issue on a GitHub repository.",
"parameters": {
"type": "object",
"properties": {
"owner": {
"type": "string",
"description": "Repository owner (user or organization)",
},
"repo": {"type": "string", "description": "Repository name"},
"title": {"type": "string", "description": "Issue title"},
"body": {"type": "string", "description": "Issue body (Markdown)"},
"labels": {
"type": "array",
"items": {"type": "string"},
"description": "Labels to apply",
},
"assignees": {
"type": "array",
"items": {"type": "string"},
"description": "Usernames to assign",
},
},
"required": ["owner", "repo", "title"],
},
"handler": github_create_issue,
},
{
"name": "github_list_issues",
"description": "List issues on a GitHub repository with optional filters for state and labels.",
"parameters": {
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
"state": {
"type": "string",
"enum": ["open", "closed", "all"],
"description": "Filter by state",
},
"labels": {
"type": "string",
"description": "Comma-separated label names to filter by",
},
"per_page": {"type": "integer", "description": "Results per page"},
"page": {"type": "integer", "description": "Page number"},
},
"required": ["owner", "repo"],
},
"handler": github_list_issues,
},
{
"name": "github_list_pull_requests",
"description": "List pull requests on a GitHub repository.",
"parameters": {
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
"state": {
"type": "string",
"enum": ["open", "closed", "all"],
"description": "Filter by state",
},
"sort": {
"type": "string",
"enum": ["created", "updated", "popularity", "long-running"],
"description": "Sort order",
},
"per_page": {"type": "integer", "description": "Results per page"},
"page": {"type": "integer", "description": "Page number"},
},
"required": ["owner", "repo"],
},
"handler": github_list_pull_requests,
},
{
"name": "github_get_pull_request",
"description": "Get details of a specific pull request, optionally including the diff.",
"parameters": {
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
"pull_number": {
"type": "integer",
"description": "Pull request number",
},
"include_diff": {
"type": "boolean",
"description": "Include the PR diff (first 8000 chars)",
},
},
"required": ["owner", "repo", "pull_number"],
},
"handler": github_get_pull_request,
},
{
"name": "github_create_gist",
"description": "Create a GitHub Gist with one or more files.",
"parameters": {
"type": "object",
"properties": {
"description": {"type": "string", "description": "Gist description"},
"files": {
"type": "object",
"description": "Map of filename -> file content",
"additionalProperties": {"type": "string"},
},
"public": {
"type": "boolean",
"description": "Whether the gist is public (default: false)",
},
},
"required": ["description", "files"],
},
"handler": github_create_gist,
},
{
"name": "github_get_file",
"description": "Get the contents of a file or directory listing from a GitHub repository.",
"parameters": {
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
"path": {
"type": "string",
"description": "File or directory path within the repo",
},
"ref": {
"type": "string",
"description": "Branch, tag, or commit SHA (default: repo default branch)",
},
},
"required": ["owner", "repo", "path"],
},
"handler": github_get_file,
},
{
"name": "github_list_notifications",
"description": "List the user's GitHub notifications (issues, PRs, releases, etc.).",
"parameters": {
"type": "object",
"properties": {
"all_notifications": {
"type": "boolean",
"description": "Show all notifications, not just unread",
},
"per_page": {"type": "integer", "description": "Results per page"},
"page": {"type": "integer", "description": "Page number"},
},
},
"handler": github_list_notifications,
},
{
"name": "github_star_repo",
"description": "Star or unstar a GitHub repository.",
"parameters": {
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
"star": {
"type": "boolean",
"description": "True to star, false to unstar (default: true)",
},
},
"required": ["owner", "repo"],
},
"handler": github_star_repo,
},
]