"""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 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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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:
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,
},
]