Compare commits

...

6 Commits

Author SHA1 Message Date
Timothy e3c71f77de chore: fix ruff format 2026-02-02 17:37:37 -08:00
Timothy b09824faec chore: fix lint 2026-02-02 17:36:02 -08:00
Lakshitaa Chellaramani 0715fc5498 Merge branch 'main' into feature/github-tool 2026-01-31 23:31:22 +05:30
lakshitaa f9fddd6663 fix(github-tool): Address PR feedback - security and integration fixes
Addresses all blockers and suggestions from code review:

**Blockers fixed:**
1. Register tools in tools/__init__.py - Added import, registration call,
   and all 13 tool names to return list
2. Add credential spec - Created GitHub entry in credentials/integrations.py
   with env_var, tools list, help URL, and health check config
3. Move tests to correct location - Relocated from
   tools/src/.../github_tool/tests/ to tools/tests/tools/test_github_tool.py
4. Removed .claude/settings.local.json from PR

**Security improvements:**
1. URL parameter sanitization - Added _sanitize_path_param() to reject
   path traversal attempts (/ or ..) in owner, repo, branch, username params
2. Error message sanitization - Added _sanitize_error_message() to prevent
   token leaks from httpx.RequestError exceptions

All 38 tests passing.
2026-01-31 23:26:33 +05:30
lakshitaa bfb660275e feat(tools): Add GitHub tool for repository and issue management
Implements comprehensive GitHub REST API v3 integration with 15 MCP tools
for managing repositories, issues, pull requests, code search, and branches.

Features:
- Repository management (list, get, search repos)
- Issue operations (create, update, close, list issues)
- Pull request management (create, list, get PRs)
- Code search across GitHub
- Branch operations (list, get branch info)

Technical details:
- 15 MCP tools organized in 5 categories
- 38 comprehensive tests with mocking (all passing)
- Full credential store support (env var + CredentialStoreAdapter)
- Proper error handling (timeout, network, API errors)
- Follows HubSpot/Slack tool patterns exactly

Files:
- tools/src/aden_tools/tools/github_tool/github_tool.py (757 lines)
- tools/src/aden_tools/tools/github_tool/tests/test_github_tool.py (628 lines)
- tools/src/aden_tools/tools/github_tool/README.md (646 lines)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-01-31 04:32:46 +05:30
lakshitaa d6ae48bc58 Merge upstream/main 2026-01-31 03:19:12 +05:30
7 changed files with 2165 additions and 10 deletions
+16 -10
View File
@@ -25,16 +25,22 @@
"Bash(xargs cat:*)",
"mcp__agent-builder__list_mcp_tools",
"mcp__agent-builder__add_mcp_server",
"mcp__agent-builder__check_missing_credentials",
"mcp__agent-builder__store_credential",
"mcp__agent-builder__list_stored_credentials",
"mcp__agent-builder__delete_stored_credential",
"mcp__agent-builder__verify_credentials",
"Bash(PYTHONPATH=/home/timothy/oss/hive/core:/home/timothy/oss/hive/exports python:*)",
"Bash(PYTHONPATH=core:exports:tools/src python -m hubspot_input:*)",
"mcp__agent-builder__export_graph"
"Bash(gh issue list:*)",
"WebFetch(domain:github.com)",
"Bash(pip install:*)",
"Bash(python -m pytest:*)",
"Bash(git checkout:*)",
"Bash(git add:*)",
"Bash(git commit -m \"$\\(cat <<''EOF''\nfeat\\(tools\\): Add Excel tool for spreadsheet operations\n\nAdds a new Excel tool for reading and manipulating .xlsx/.xlsm files:\n- excel_read: Read Excel files with pagination and sheet selection\n- excel_write: Create new Excel files with data\n- excel_append: Append rows to existing files\n- excel_info: Get metadata about Excel files \\(sheets, columns, row counts\\)\n- excel_sheet_list: List all sheets in a workbook\n\nIncludes comprehensive test coverage \\(37 tests\\) and documentation.\n\nReferences #2805\n\nCo-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>\nEOF\n\\)\")",
"Bash(git push:*)",
"Bash(git pull:*)",
"Bash(git stash:*)",
"Bash(git merge:*)"
]
},
"enabledMcpjsonServers": ["agent-builder", "tools"],
"enableAllProjectMcpServers": true
"enableAllProjectMcpServers": true,
"enabledMcpjsonServers": [
"agent-builder",
"tools"
]
}
@@ -7,6 +7,47 @@ Contains credentials for third-party service integrations (HubSpot, etc.).
from .base import CredentialSpec
INTEGRATION_CREDENTIALS = {
"github": CredentialSpec(
env_var="GITHUB_TOKEN",
tools=[
"github_list_repos",
"github_get_repo",
"github_search_repos",
"github_list_issues",
"github_get_issue",
"github_create_issue",
"github_update_issue",
"github_list_pull_requests",
"github_get_pull_request",
"github_create_pull_request",
"github_search_code",
"github_list_branches",
"github_get_branch",
],
required=True,
startup_required=False,
help_url="https://github.com/settings/tokens",
description="GitHub Personal Access Token (classic)",
# Auth method support
aden_supported=False,
direct_api_key_supported=True,
api_key_instructions="""To get a GitHub Personal Access Token:
1. Go to GitHub Settings > Developer settings > Personal access tokens
2. Click "Generate new token" > "Generate new token (classic)"
3. Give your token a descriptive name (e.g., "Hive Agent")
4. Select the following scopes:
- repo (Full control of private repositories)
- read:org (Read org and team membership - optional)
- user (Read user profile data - optional)
5. Click "Generate token" and copy the token (starts with ghp_)
6. Store it securely - you won't be able to see it again!""",
# Health check configuration
health_check_endpoint="https://api.github.com/user",
health_check_method="GET",
# Credential store mapping
credential_id="github",
credential_key="access_token",
),
"hubspot": CredentialSpec(
env_var="HUBSPOT_ACCESS_TOKEN",
tools=[
+15
View File
@@ -38,6 +38,7 @@ from .file_system_toolkits.replace_file_content import (
# Import file system toolkits
from .file_system_toolkits.view_file import register_tools as register_view_file
from .file_system_toolkits.write_to_file import register_tools as register_write_to_file
from .github_tool import register_tools as register_github
from .hubspot_tool import register_tools as register_hubspot
from .pdf_read_tool import register_tools as register_pdf_read
from .web_scrape_tool import register_tools as register_web_scrape
@@ -67,6 +68,7 @@ def register_all_tools(
# Tools that need credentials (pass credentials if provided)
# web_search supports multiple providers (Google, Brave) with auto-detection
register_web_search(mcp, credentials=credentials)
register_github(mcp, credentials=credentials)
# email supports multiple providers (Resend) with auto-detection
register_email(mcp, credentials=credentials)
register_hubspot(mcp, credentials=credentials)
@@ -100,6 +102,19 @@ def register_all_tools(
"csv_append",
"csv_info",
"csv_sql",
"github_list_repos",
"github_get_repo",
"github_search_repos",
"github_list_issues",
"github_get_issue",
"github_create_issue",
"github_update_issue",
"github_list_pull_requests",
"github_get_pull_request",
"github_create_pull_request",
"github_search_code",
"github_list_branches",
"github_get_branch",
"send_email",
"send_budget_alert_email",
"hubspot_search_contacts",
@@ -0,0 +1,646 @@
# GitHub Tool
Interact with GitHub repositories, issues, and pull requests within the Aden agent framework.
## Installation
The GitHub tool uses `httpx` which is already included in the base dependencies. No additional installation required.
## Setup
You need a GitHub Personal Access Token (PAT) to use this tool.
### Getting a GitHub Token
1. Go to https://github.com/settings/tokens
2. Click "Generate new token" → "Generate new token (classic)"
3. Give your token a descriptive name (e.g., "Aden Agent Framework")
4. Select the following scopes:
- `repo` - Full control of private repositories (includes all repo scopes)
- `read:org` - Read org and team membership (optional, for org access)
- `user` - Read user profile data (optional)
5. Click "Generate token"
6. Copy the token (starts with `ghp_`)
**Note:** Keep your token secure! It provides access to your GitHub account.
### Configuration
Set the token as an environment variable:
```bash
export GITHUB_TOKEN=ghp_your_token_here
```
Or configure via the credential store (recommended for production).
## Available Functions
### Repository Management
#### `github_list_repos`
List repositories for a user or the authenticated user.
**Parameters:**
- `username` (str, optional): GitHub username (if None, lists authenticated user's repos)
- `visibility` (str, optional): Repository visibility ("all", "public", "private", default "all")
- `sort` (str, optional): Sort order ("created", "updated", "pushed", "full_name", default "updated")
- `limit` (int, optional): Maximum number of repositories (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": [
{
"id": 123456,
"name": "my-repo",
"full_name": "username/my-repo",
"description": "A cool project",
"private": False,
"html_url": "https://github.com/username/my-repo",
"stargazers_count": 42,
"forks_count": 7
}
]
}
```
**Example:**
```python
# List your repositories
result = github_list_repos()
# List another user's public repositories
result = github_list_repos(username="octocat", limit=10)
```
#### `github_get_repo`
Get detailed information about a specific repository.
**Parameters:**
- `owner` (str): Repository owner (username or organization)
- `repo` (str): Repository name
**Returns:**
```python
{
"success": True,
"data": {
"id": 123456,
"name": "my-repo",
"full_name": "owner/my-repo",
"description": "Project description",
"private": False,
"default_branch": "main",
"stargazers_count": 100,
"forks_count": 25,
"language": "Python",
"created_at": "2024-01-01T00:00:00Z",
"updated_at": "2024-01-31T12:00:00Z"
}
}
```
**Example:**
```python
result = github_get_repo(owner="adenhq", repo="hive")
print(f"Stars: {result['data']['stargazers_count']}")
```
#### `github_search_repos`
Search for repositories on GitHub.
**Parameters:**
- `query` (str): Search query (supports GitHub search syntax)
- `sort` (str, optional): Sort field ("stars", "forks", "updated")
- `limit` (int, optional): Maximum results (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": {
"total_count": 1000,
"items": [
{
"id": 123,
"name": "awesome-python",
"full_name": "user/awesome-python",
"description": "A curated list",
"stargazers_count": 5000
}
]
}
}
```
**Example:**
```python
# Search for Python repos with many stars
result = github_search_repos(
query="language:python stars:>1000",
sort="stars",
limit=10
)
# Search in a specific organization
result = github_search_repos(query="org:adenhq agent")
```
### Issue Management
#### `github_list_issues`
List issues for a repository.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `state` (str, optional): Issue state ("open", "closed", "all", default "open")
- `limit` (int, optional): Maximum issues (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": [
{
"number": 42,
"title": "Bug in feature X",
"state": "open",
"user": {"login": "username"},
"labels": [{"name": "bug"}],
"created_at": "2024-01-30T10:00:00Z",
"html_url": "https://github.com/owner/repo/issues/42"
}
]
}
```
**Example:**
```python
# List open issues
issues = github_list_issues(owner="adenhq", repo="hive", state="open")
for issue in issues["data"]:
print(f"#{issue['number']}: {issue['title']}")
```
#### `github_get_issue`
Get a specific issue by number.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `issue_number` (int): Issue number
**Returns:**
```python
{
"success": True,
"data": {
"number": 42,
"title": "Issue title",
"body": "Detailed description...",
"state": "open",
"user": {"login": "username"},
"assignees": [],
"labels": [{"name": "enhancement"}],
"comments": 5
}
}
```
**Example:**
```python
issue = github_get_issue(owner="adenhq", repo="hive", issue_number=2805)
print(issue["data"]["body"])
```
#### `github_create_issue`
Create a new issue in a repository.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `title` (str): Issue title
- `body` (str, optional): Issue description (supports Markdown)
- `labels` (list[str], optional): List of label names
- `assignees` (list[str], optional): List of usernames to assign
**Returns:**
```python
{
"success": True,
"data": {
"number": 43,
"title": "New issue",
"html_url": "https://github.com/owner/repo/issues/43"
}
}
```
**Example:**
```python
result = github_create_issue(
owner="myorg",
repo="myrepo",
title="Add new feature",
body="## Description\n\nWe need to add...",
labels=["enhancement", "help wanted"],
assignees=["developer1"]
)
print(f"Created issue #{result['data']['number']}")
```
#### `github_update_issue`
Update an existing issue.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `issue_number` (int): Issue number
- `title` (str, optional): New title
- `body` (str, optional): New body
- `state` (str, optional): New state ("open" or "closed")
- `labels` (list[str], optional): New list of label names
**Returns:**
```python
{
"success": True,
"data": {
"number": 43,
"title": "Updated title",
"state": "closed"
}
}
```
**Example:**
```python
# Close an issue
result = github_update_issue(
owner="myorg",
repo="myrepo",
issue_number=43,
state="closed",
body="Fixed in PR #44"
)
```
### Pull Request Management
#### `github_list_pull_requests`
List pull requests for a repository.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `state` (str, optional): PR state ("open", "closed", "all", default "open")
- `limit` (int, optional): Maximum PRs (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": [
{
"number": 10,
"title": "Add new feature",
"state": "open",
"user": {"login": "contributor"},
"head": {"ref": "feature-branch"},
"base": {"ref": "main"},
"html_url": "https://github.com/owner/repo/pull/10"
}
]
}
```
**Example:**
```python
prs = github_list_pull_requests(owner="adenhq", repo="hive", state="open")
for pr in prs["data"]:
print(f"PR #{pr['number']}: {pr['title']}")
```
#### `github_get_pull_request`
Get a specific pull request.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `pull_number` (int): Pull request number
**Returns:**
```python
{
"success": True,
"data": {
"number": 10,
"title": "PR title",
"body": "Description...",
"state": "open",
"merged": False,
"draft": False,
"head": {"ref": "feature"},
"base": {"ref": "main"}
}
}
```
**Example:**
```python
pr = github_get_pull_request(owner="adenhq", repo="hive", pull_number=2814)
print(f"PR by {pr['data']['user']['login']}")
```
#### `github_create_pull_request`
Create a new pull request.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `title` (str): Pull request title
- `head` (str): Branch with your changes (e.g., "my-feature")
- `base` (str): Branch to merge into (e.g., "main")
- `body` (str, optional): Pull request description (supports Markdown)
- `draft` (bool, optional): Create as draft PR (default False)
**Returns:**
```python
{
"success": True,
"data": {
"number": 11,
"title": "New PR",
"html_url": "https://github.com/owner/repo/pull/11"
}
}
```
**Example:**
```python
result = github_create_pull_request(
owner="myorg",
repo="myrepo",
title="feat: Add GitHub integration tool",
head="feature/github-tool",
base="main",
body="## Summary\n\n- Implements GitHub API integration\n- Adds 30+ tests",
draft=False
)
print(f"Created PR: {result['data']['html_url']}")
```
### Search
#### `github_search_code`
Search code across GitHub.
**Parameters:**
- `query` (str): Search query (supports GitHub code search syntax)
- `limit` (int, optional): Maximum results (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": {
"total_count": 50,
"items": [
{
"name": "example.py",
"path": "src/example.py",
"repository": {
"full_name": "owner/repo"
},
"html_url": "https://github.com/owner/repo/blob/main/src/example.py"
}
]
}
}
```
**Example:**
```python
# Search for function usage
result = github_search_code(
query="register_tools language:python repo:adenhq/hive"
)
# Search for specific code pattern
result = github_search_code(query="FastMCP extension:py")
```
### Branch Management
#### `github_list_branches`
List branches for a repository.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `limit` (int, optional): Maximum branches (1-100, default 30)
**Returns:**
```python
{
"success": True,
"data": [
{
"name": "main",
"protected": True,
"commit": {"sha": "abc123..."}
},
{
"name": "develop",
"protected": False
}
]
}
```
**Example:**
```python
branches = github_list_branches(owner="adenhq", repo="hive")
for branch in branches["data"]:
print(f"Branch: {branch['name']}")
```
#### `github_get_branch`
Get information about a specific branch.
**Parameters:**
- `owner` (str): Repository owner
- `repo` (str): Repository name
- `branch` (str): Branch name
**Returns:**
```python
{
"success": True,
"data": {
"name": "main",
"protected": True,
"commit": {
"sha": "abc123...",
"commit": {
"message": "Latest commit message"
}
}
}
}
```
**Example:**
```python
main_branch = github_get_branch(owner="adenhq", repo="hive", branch="main")
print(f"Latest commit: {main_branch['data']['commit']['sha']}")
```
## Error Handling
All functions return a dict with an `error` key if something goes wrong:
```python
{
"error": "GitHub API error (HTTP 404): Not Found"
}
```
Common errors:
- `not configured` - No GitHub token provided
- `Invalid or expired GitHub token` - Token authentication failed (401)
- `Forbidden` - Insufficient permissions or rate limit exceeded (403)
- `Resource not found` - Repository, issue, or PR doesn't exist (404)
- `Validation error` - Invalid request parameters (422)
- `Request timed out` - Network timeout
- `Network error` - Connection issues
## Security
- Personal Access Tokens are never logged or exposed
- All API calls use HTTPS
- Tokens are retrieved from secure credential store or environment variables
- Fine-grained permissions can be configured via GitHub token scopes
## Use Cases
### Automated Issue Management
```python
# Create issues from bug reports
github_create_issue(
owner="myorg",
repo="myapp",
title="Bug: Login fails on mobile",
body="## Steps to reproduce\n1. Open app on mobile...",
labels=["bug", "mobile"]
)
```
### CI/CD Integration
```python
# Create PR after automated changes
github_create_pull_request(
owner="myorg",
repo="myrepo",
title="chore: Update dependencies",
head="bot/update-deps",
base="main",
body="Automated dependency updates"
)
```
### Repository Analytics
```python
# Analyze repository activity
repo = github_get_repo(owner="adenhq", repo="hive")
issues = github_list_issues(owner="adenhq", repo="hive", state="open")
prs = github_list_pull_requests(owner="adenhq", repo="hive", state="open")
print(f"Stars: {repo['data']['stargazers_count']}")
print(f"Open Issues: {len(issues['data'])}")
print(f"Open PRs: {len(prs['data'])}")
```
### Code Discovery
```python
# Find examples of API usage
results = github_search_code(
query="register_tools language:python",
limit=50
)
for item in results["data"]["items"]:
print(f"Found in: {item['repository']['full_name']}")
```
### Project Automation
```python
# Auto-close stale issues
issues = github_list_issues(owner="myorg", repo="myrepo", state="open")
for issue in issues["data"]:
# Check if stale (custom logic)
if is_stale(issue):
github_update_issue(
owner="myorg",
repo="myrepo",
issue_number=issue["number"],
state="closed",
body="Closing due to inactivity"
)
```
## Rate Limits
GitHub enforces rate limits on API calls:
- **Authenticated requests**: 5,000 requests per hour
- **Search API**: 30 requests per minute
- **Unauthenticated requests**: 60 requests per hour (not applicable with token)
The tool handles rate limit errors gracefully with appropriate error messages. Monitor your usage at: https://api.github.com/rate_limit
## GitHub Search Syntax
For `github_search_repos` and `github_search_code`, you can use advanced search qualifiers:
### Repository Search
- `language:python` - Filter by language
- `stars:>1000` - Repositories with more than 1000 stars
- `forks:>100` - Repositories with more than 100 forks
- `org:adenhq` - Search within an organization
- `topic:machine-learning` - Filter by topic
- `created:>2024-01-01` - Created after date
### Code Search
- `repo:owner/repo` - Search in specific repository
- `extension:py` - Filter by file extension
- `path:src/` - Search in specific path
- `language:python` - Filter by language
Examples:
```python
# Find popular Python ML projects
github_search_repos(
query="language:python topic:machine-learning stars:>5000",
sort="stars"
)
# Find FastMCP usage examples
github_search_code(
query="FastMCP extension:py"
)
```
@@ -0,0 +1,5 @@
"""GitHub Tool package."""
from .github_tool import register_tools
__all__ = ["register_tools"]
@@ -0,0 +1,818 @@
"""
GitHub Tool - Interact with GitHub repositories, issues, and pull requests.
Supports:
- Personal Access Tokens (GITHUB_TOKEN / ghp_...)
- OAuth tokens via the credential store
API Reference: https://docs.github.com/en/rest
"""
from __future__ import annotations
import os
from typing import TYPE_CHECKING, Any
import httpx
from fastmcp import FastMCP
if TYPE_CHECKING:
from aden_tools.credentials import CredentialStoreAdapter
GITHUB_API_BASE = "https://api.github.com"
def _sanitize_path_param(param: str, param_name: str = "parameter") -> str:
"""
Sanitize URL path parameters to prevent path traversal.
Args:
param: The parameter value to sanitize
param_name: Name of the parameter (for error messages)
Returns:
The sanitized parameter
Raises:
ValueError: If parameter contains invalid characters
"""
if "/" in param or ".." in param:
raise ValueError(f"Invalid {param_name}: cannot contain '/' or '..'")
return param
def _sanitize_error_message(error: Exception) -> str:
"""
Sanitize error messages to prevent token leaks.
httpx.RequestError can include headers in the exception message,
which may expose the Bearer token.
Args:
error: The exception to sanitize
Returns:
A safe error message without sensitive information
"""
error_str = str(error)
# Remove any Authorization headers or Bearer tokens
if "Authorization" in error_str or "Bearer" in error_str:
return "Network error occurred"
return f"Network error: {error_str}"
class _GitHubClient:
"""Internal client wrapping GitHub REST API v3 calls."""
def __init__(self, token: str):
self._token = token
@property
def _headers(self) -> dict[str, str]:
return {
"Authorization": f"Bearer {self._token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
def _handle_response(self, response: httpx.Response) -> dict[str, Any]:
"""Handle GitHub API response format."""
if response.status_code == 401:
return {"error": "Invalid or expired GitHub token"}
if response.status_code == 403:
return {"error": "Forbidden - check token permissions or rate limit"}
if response.status_code == 404:
return {"error": "Resource not found"}
if response.status_code == 422:
try:
detail = response.json().get("message", "Validation failed")
except Exception:
detail = "Validation failed"
return {"error": f"Validation error: {detail}"}
if response.status_code >= 400:
try:
detail = response.json().get("message", response.text)
except Exception:
detail = response.text
return {"error": f"GitHub API error (HTTP {response.status_code}): {detail}"}
try:
return {"success": True, "data": response.json()}
except Exception:
return {"success": True, "data": {}}
# --- Repositories ---
def list_repos(
self,
username: str | None = None,
visibility: str = "all",
sort: str = "updated",
limit: int = 30,
) -> dict[str, Any]:
"""List repositories for a user or authenticated user."""
if username:
username = _sanitize_path_param(username, "username")
url = f"{GITHUB_API_BASE}/users/{username}/repos"
else:
url = f"{GITHUB_API_BASE}/user/repos"
params = {
"visibility": visibility,
"sort": sort,
"per_page": min(limit, 100),
}
response = httpx.get(
url,
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
def get_repo(
self,
owner: str,
repo: str,
) -> dict[str, Any]:
"""Get repository information."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def search_repos(
self,
query: str,
sort: str | None = None,
limit: int = 30,
) -> dict[str, Any]:
"""Search for repositories."""
params: dict[str, Any] = {
"q": query,
"per_page": min(limit, 100),
}
if sort:
params["sort"] = sort
response = httpx.get(
f"{GITHUB_API_BASE}/search/repositories",
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
# --- Issues ---
def list_issues(
self,
owner: str,
repo: str,
state: str = "open",
limit: int = 30,
) -> dict[str, Any]:
"""List issues for a repository."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
params = {
"state": state,
"per_page": min(limit, 100),
}
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/issues",
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
def get_issue(
self,
owner: str,
repo: str,
issue_number: int,
) -> dict[str, Any]:
"""Get a specific issue."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/issues/{issue_number}",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def create_issue(
self,
owner: str,
repo: str,
title: str,
body: str | None = None,
labels: list[str] | None = None,
assignees: list[str] | None = None,
) -> dict[str, Any]:
"""Create a new issue."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
payload: dict[str, Any] = {"title": title}
if body:
payload["body"] = body
if labels:
payload["labels"] = labels
if assignees:
payload["assignees"] = assignees
response = httpx.post(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/issues",
headers=self._headers,
json=payload,
timeout=30.0,
)
return self._handle_response(response)
def update_issue(
self,
owner: str,
repo: str,
issue_number: int,
title: str | None = None,
body: str | None = None,
state: str | None = None,
labels: list[str] | None = None,
) -> dict[str, Any]:
"""Update an existing issue."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
payload: dict[str, Any] = {}
if title:
payload["title"] = title
if body is not None:
payload["body"] = body
if state:
payload["state"] = state
if labels is not None:
payload["labels"] = labels
response = httpx.patch(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/issues/{issue_number}",
headers=self._headers,
json=payload,
timeout=30.0,
)
return self._handle_response(response)
# --- Pull Requests ---
def list_pull_requests(
self,
owner: str,
repo: str,
state: str = "open",
limit: int = 30,
) -> dict[str, Any]:
"""List pull requests for a repository."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
params = {
"state": state,
"per_page": min(limit, 100),
}
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/pulls",
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
def get_pull_request(
self,
owner: str,
repo: str,
pull_number: int,
) -> dict[str, Any]:
"""Get a specific pull request."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/pulls/{pull_number}",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def create_pull_request(
self,
owner: str,
repo: str,
title: str,
head: str,
base: str,
body: str | None = None,
draft: bool = False,
) -> dict[str, Any]:
"""Create a new pull request."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
payload: dict[str, Any] = {
"title": title,
"head": head,
"base": base,
"draft": draft,
}
if body:
payload["body"] = body
response = httpx.post(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/pulls",
headers=self._headers,
json=payload,
timeout=30.0,
)
return self._handle_response(response)
# --- Search ---
def search_code(
self,
query: str,
limit: int = 30,
) -> dict[str, Any]:
"""Search code across GitHub."""
params = {
"q": query,
"per_page": min(limit, 100),
}
response = httpx.get(
f"{GITHUB_API_BASE}/search/code",
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
# --- Branches ---
def list_branches(
self,
owner: str,
repo: str,
limit: int = 30,
) -> dict[str, Any]:
"""List branches for a repository."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
params = {
"per_page": min(limit, 100),
}
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/branches",
headers=self._headers,
params=params,
timeout=30.0,
)
return self._handle_response(response)
def get_branch(
self,
owner: str,
repo: str,
branch: str,
) -> dict[str, Any]:
"""Get a specific branch."""
owner = _sanitize_path_param(owner, "owner")
repo = _sanitize_path_param(repo, "repo")
branch = _sanitize_path_param(branch, "branch")
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/branches/{branch}",
headers=self._headers,
timeout=30.0,
)
return self._handle_response(response)
def register_tools(
mcp: FastMCP,
credentials: CredentialStoreAdapter | None = None,
) -> None:
"""Register GitHub tools with the MCP server."""
def _get_token() -> str | None:
"""Get GitHub token from credential manager or environment."""
if credentials is not None:
token = credentials.get("github")
if token is not None and not isinstance(token, str):
raise TypeError(
f"Expected string from credentials.get('github'), got {type(token).__name__}"
)
return token
return os.getenv("GITHUB_TOKEN")
def _get_client() -> _GitHubClient | dict[str, str]:
"""Get a GitHub client, or return an error dict if no credentials."""
token = _get_token()
if not token:
return {
"error": "GitHub credentials not configured",
"help": (
"Set GITHUB_TOKEN environment variable "
"or configure via credential store. "
"Get a token at https://github.com/settings/tokens"
),
}
return _GitHubClient(token)
# --- Repositories ---
@mcp.tool()
def github_list_repos(
username: str | None = None,
visibility: str = "all",
sort: str = "updated",
limit: int = 30,
) -> dict:
"""
List repositories for a user or the authenticated user.
Args:
username: GitHub username (if None, lists authenticated user's repos)
visibility: Repository visibility filter ("all", "public", "private")
sort: Sort order ("created", "updated", "pushed", "full_name")
limit: Maximum number of repositories to return (1-100, default 30)
Returns:
Dict with list of repositories or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.list_repos(username, visibility, sort, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_get_repo(
owner: str,
repo: str,
) -> dict:
"""
Get information about a specific repository.
Args:
owner: Repository owner (username or organization)
repo: Repository name
Returns:
Dict with repository information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_repo(owner, repo)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_search_repos(
query: str,
sort: str | None = None,
limit: int = 30,
) -> dict:
"""
Search for repositories on GitHub.
Args:
query: Search query (e.g., "language:python stars:>1000")
sort: Sort field ("stars", "forks", "updated")
limit: Maximum number of results (1-100, default 30)
Returns:
Dict with search results or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.search_repos(query, sort, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
# --- Issues ---
@mcp.tool()
def github_list_issues(
owner: str,
repo: str,
state: str = "open",
limit: int = 30,
) -> dict:
"""
List issues for a repository.
Args:
owner: Repository owner
repo: Repository name
state: Issue state ("open", "closed", "all")
limit: Maximum number of issues to return (1-100, default 30)
Returns:
Dict with list of issues or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.list_issues(owner, repo, state, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_get_issue(
owner: str,
repo: str,
issue_number: int,
) -> dict:
"""
Get a specific issue.
Args:
owner: Repository owner
repo: Repository name
issue_number: Issue number
Returns:
Dict with issue information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_issue(owner, repo, issue_number)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_create_issue(
owner: str,
repo: str,
title: str,
body: str | None = None,
labels: list[str] | None = None,
assignees: list[str] | None = None,
) -> dict:
"""
Create a new issue in a repository.
Args:
owner: Repository owner
repo: Repository name
title: Issue title
body: Issue body/description (supports Markdown)
labels: List of label names to apply
assignees: List of usernames to assign
Returns:
Dict with created issue information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.create_issue(owner, repo, title, body, labels, assignees)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_update_issue(
owner: str,
repo: str,
issue_number: int,
title: str | None = None,
body: str | None = None,
state: str | None = None,
labels: list[str] | None = None,
) -> dict:
"""
Update an existing issue.
Args:
owner: Repository owner
repo: Repository name
issue_number: Issue number
title: New issue title
body: New issue body
state: New state ("open" or "closed")
labels: New list of label names
Returns:
Dict with updated issue information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.update_issue(owner, repo, issue_number, title, body, state, labels)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
# --- Pull Requests ---
@mcp.tool()
def github_list_pull_requests(
owner: str,
repo: str,
state: str = "open",
limit: int = 30,
) -> dict:
"""
List pull requests for a repository.
Args:
owner: Repository owner
repo: Repository name
state: PR state ("open", "closed", "all")
limit: Maximum number of PRs to return (1-100, default 30)
Returns:
Dict with list of pull requests or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.list_pull_requests(owner, repo, state, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_get_pull_request(
owner: str,
repo: str,
pull_number: int,
) -> dict:
"""
Get a specific pull request.
Args:
owner: Repository owner
repo: Repository name
pull_number: Pull request number
Returns:
Dict with pull request information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_pull_request(owner, repo, pull_number)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_create_pull_request(
owner: str,
repo: str,
title: str,
head: str,
base: str,
body: str | None = None,
draft: bool = False,
) -> dict:
"""
Create a new pull request.
Args:
owner: Repository owner
repo: Repository name
title: Pull request title
head: The name of the branch where your changes are (e.g., "my-feature")
base: The name of the branch you want to merge into (e.g., "main")
body: Pull request description (supports Markdown)
draft: Whether to create as a draft PR
Returns:
Dict with created pull request information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.create_pull_request(owner, repo, title, head, base, body, draft)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
# --- Search ---
@mcp.tool()
def github_search_code(
query: str,
limit: int = 30,
) -> dict:
"""
Search code across GitHub.
Args:
query: Search query (e.g., "addClass repo:jquery/jquery")
limit: Maximum number of results (1-100, default 30)
Returns:
Dict with search results or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.search_code(query, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
# --- Branches ---
@mcp.tool()
def github_list_branches(
owner: str,
repo: str,
limit: int = 30,
) -> dict:
"""
List branches for a repository.
Args:
owner: Repository owner
repo: Repository name
limit: Maximum number of branches to return (1-100, default 30)
Returns:
Dict with list of branches or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.list_branches(owner, repo, limit)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
@mcp.tool()
def github_get_branch(
owner: str,
repo: str,
branch: str,
) -> dict:
"""
Get information about a specific branch.
Args:
owner: Repository owner
repo: Repository name
branch: Branch name
Returns:
Dict with branch information or error
"""
client = _get_client()
if isinstance(client, dict):
return client
try:
return client.get_branch(owner, repo, branch)
except httpx.TimeoutException:
return {"error": "Request timed out"}
except httpx.RequestError as e:
return {"error": _sanitize_error_message(e)}
+624
View File
@@ -0,0 +1,624 @@
"""
Tests for GitHub tool.
Covers:
- _GitHubClient methods (repositories, issues, PRs, search, branches)
- Error handling (API errors, timeout, network errors)
- Credential retrieval (CredentialStoreAdapter vs env var)
- All 15 MCP tool functions
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import httpx
import pytest
from fastmcp import FastMCP
from aden_tools.tools.github_tool.github_tool import (
_GitHubClient,
register_tools,
)
# --- _GitHubClient tests ---
class TestGitHubClient:
def setup_method(self):
self.client = _GitHubClient("ghp_test_token")
def test_headers(self):
headers = self.client._headers
assert headers["Authorization"] == "Bearer ghp_test_token"
assert "application/vnd.github+json" in headers["Accept"]
def test_handle_response_success(self):
response = MagicMock()
response.status_code = 200
response.json.return_value = {"id": 123, "name": "test-repo"}
result = self.client._handle_response(response)
assert result["success"] is True
assert result["data"]["name"] == "test-repo"
def test_handle_response_401(self):
response = MagicMock()
response.status_code = 401
result = self.client._handle_response(response)
assert "error" in result
assert "Invalid or expired" in result["error"]
def test_handle_response_403(self):
response = MagicMock()
response.status_code = 403
result = self.client._handle_response(response)
assert "error" in result
assert "Forbidden" in result["error"]
def test_handle_response_404(self):
response = MagicMock()
response.status_code = 404
result = self.client._handle_response(response)
assert "error" in result
assert "not found" in result["error"]
def test_handle_response_422(self):
response = MagicMock()
response.status_code = 422
response.json.return_value = {"message": "Validation failed"}
result = self.client._handle_response(response)
assert "error" in result
assert "Validation" in result["error"]
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_repos(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [
{"id": 1, "name": "repo1", "full_name": "user/repo1"},
{"id": 2, "name": "repo2", "full_name": "user/repo2"},
]
mock_get.return_value = mock_response
result = self.client.list_repos(username="testuser")
mock_get.assert_called_once()
assert result["success"] is True
assert len(result["data"]) == 2
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_repos_authenticated_user(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = []
mock_get.return_value = mock_response
self.client.list_repos(username=None)
call_url = mock_get.call_args.args[0]
assert "/user/repos" in call_url
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_repo(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"id": 123,
"name": "test-repo",
"full_name": "owner/test-repo",
"description": "A test repository",
}
mock_get.return_value = mock_response
result = self.client.get_repo("owner", "test-repo")
assert result["success"] is True
assert result["data"]["name"] == "test-repo"
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_search_repos(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"total_count": 1,
"items": [{"id": 123, "name": "test-repo"}],
}
mock_get.return_value = mock_response
result = self.client.search_repos("language:python")
assert result["success"] is True
assert "items" in result["data"]
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_issues(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [
{"number": 1, "title": "Issue 1", "state": "open"},
{"number": 2, "title": "Issue 2", "state": "open"},
]
mock_get.return_value = mock_response
result = self.client.list_issues("owner", "repo", state="open")
assert result["success"] is True
assert len(result["data"]) == 2
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_issue(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"number": 1,
"title": "Test Issue",
"body": "This is a test",
}
mock_get.return_value = mock_response
result = self.client.get_issue("owner", "repo", 1)
assert result["success"] is True
assert result["data"]["title"] == "Test Issue"
@patch("aden_tools.tools.github_tool.github_tool.httpx.post")
def test_create_issue(self, mock_post):
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.json.return_value = {
"number": 42,
"title": "New Issue",
"body": "Description",
}
mock_post.return_value = mock_response
result = self.client.create_issue(
"owner", "repo", "New Issue", body="Description", labels=["bug"]
)
assert result["success"] is True
assert result["data"]["number"] == 42
call_json = mock_post.call_args.kwargs["json"]
assert call_json["labels"] == ["bug"]
@patch("aden_tools.tools.github_tool.github_tool.httpx.patch")
def test_update_issue(self, mock_patch):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"number": 1,
"title": "Updated Title",
"state": "closed",
}
mock_patch.return_value = mock_response
result = self.client.update_issue("owner", "repo", 1, title="Updated Title", state="closed")
assert result["success"] is True
assert result["data"]["state"] == "closed"
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_pull_requests(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [
{"number": 1, "title": "PR 1", "state": "open"},
]
mock_get.return_value = mock_response
result = self.client.list_pull_requests("owner", "repo")
assert result["success"] is True
assert len(result["data"]) == 1
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_pull_request(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"number": 1,
"title": "Test PR",
"head": {"ref": "feature"},
"base": {"ref": "main"},
}
mock_get.return_value = mock_response
result = self.client.get_pull_request("owner", "repo", 1)
assert result["success"] is True
assert result["data"]["title"] == "Test PR"
@patch("aden_tools.tools.github_tool.github_tool.httpx.post")
def test_create_pull_request(self, mock_post):
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.json.return_value = {
"number": 10,
"title": "New PR",
"draft": False,
}
mock_post.return_value = mock_response
result = self.client.create_pull_request(
"owner", "repo", "New PR", "feature-branch", "main", body="PR description"
)
assert result["success"] is True
assert result["data"]["number"] == 10
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_search_code(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"total_count": 5,
"items": [{"name": "file.py", "path": "src/file.py"}],
}
mock_get.return_value = mock_response
result = self.client.search_code("addClass repo:jquery/jquery")
assert result["success"] is True
assert "items" in result["data"]
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_branches(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [
{"name": "main", "protected": True},
{"name": "develop", "protected": False},
]
mock_get.return_value = mock_response
result = self.client.list_branches("owner", "repo")
assert result["success"] is True
assert len(result["data"]) == 2
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_branch(self, mock_get):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"name": "main",
"protected": True,
"commit": {"sha": "abc123"},
}
mock_get.return_value = mock_response
result = self.client.get_branch("owner", "repo", "main")
assert result["success"] is True
assert result["data"]["name"] == "main"
# --- Credential retrieval tests ---
class TestCredentialRetrieval:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
def test_no_credentials_returns_error(self, mcp):
"""When no credentials are configured, tools return helpful error."""
with patch.dict("os.environ", {}, clear=True):
with patch("os.getenv", return_value=None):
register_tools(mcp, credentials=None)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
result = list_repos()
assert "error" in result
assert "not configured" in result["error"]
assert "help" in result
def test_env_var_token(self, mcp):
"""Token from GITHUB_TOKEN env var is used."""
with patch("os.getenv", return_value="ghp_env_token"):
with patch("aden_tools.tools.github_tool.github_tool.httpx.get") as mock_get:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = []
mock_get.return_value = mock_response
register_tools(mcp, credentials=None)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
list_repos()
call_headers = mock_get.call_args.kwargs["headers"]
assert call_headers["Authorization"] == "Bearer ghp_env_token"
def test_credential_store_token(self, mcp):
"""Token from CredentialStoreAdapter is preferred."""
mock_credentials = MagicMock()
mock_credentials.get.return_value = "ghp_store_token"
with patch("aden_tools.tools.github_tool.github_tool.httpx.get") as mock_get:
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = []
mock_get.return_value = mock_response
register_tools(mcp, credentials=mock_credentials)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
list_repos()
mock_credentials.get.assert_called_with("github")
call_headers = mock_get.call_args.kwargs["headers"]
assert call_headers["Authorization"] == "Bearer ghp_store_token"
# --- MCP Tool function tests ---
class TestGitHubListRepos:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_repos_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [{"id": 1, "name": "test-repo"}]
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
result = list_repos(username="testuser")
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_repos_timeout(self, mock_get, mcp):
mock_get.side_effect = httpx.TimeoutException("Timeout")
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
result = list_repos()
assert "error" in result
assert "timed out" in result["error"]
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_repos_network_error(self, mock_get, mcp):
mock_get.side_effect = httpx.RequestError("Network error")
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_repos = mcp._tool_manager._tools["github_list_repos"].fn
result = list_repos()
assert "error" in result
assert "Network error" in result["error"]
class TestGitHubGetRepo:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_repo_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"id": 1, "name": "test-repo"}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
get_repo = mcp._tool_manager._tools["github_get_repo"].fn
result = get_repo(owner="owner", repo="test-repo")
assert result["success"] is True
class TestGitHubSearchRepos:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_search_repos_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"total_count": 1, "items": []}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
search_repos = mcp._tool_manager._tools["github_search_repos"].fn
result = search_repos(query="python")
assert result["success"] is True
class TestGitHubIssues:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_issues_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [{"number": 1, "title": "Test Issue"}]
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_issues = mcp._tool_manager._tools["github_list_issues"].fn
result = list_issues(owner="owner", repo="repo")
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_issue_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"number": 1, "title": "Test"}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
get_issue = mcp._tool_manager._tools["github_get_issue"].fn
result = get_issue(owner="owner", repo="repo", issue_number=1)
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.post")
def test_create_issue_success(self, mock_post, mcp):
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.json.return_value = {"number": 1, "title": "New Issue"}
mock_post.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
create_issue = mcp._tool_manager._tools["github_create_issue"].fn
result = create_issue(owner="owner", repo="repo", title="New Issue")
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.patch")
def test_update_issue_success(self, mock_patch, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"number": 1, "state": "closed"}
mock_patch.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
update_issue = mcp._tool_manager._tools["github_update_issue"].fn
result = update_issue(owner="owner", repo="repo", issue_number=1, state="closed")
assert result["success"] is True
class TestGitHubPullRequests:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_pull_requests_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [{"number": 1, "title": "Test PR"}]
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_prs = mcp._tool_manager._tools["github_list_pull_requests"].fn
result = list_prs(owner="owner", repo="repo")
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_pull_request_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"number": 1, "title": "PR"}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
get_pr = mcp._tool_manager._tools["github_get_pull_request"].fn
result = get_pr(owner="owner", repo="repo", pull_number=1)
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.post")
def test_create_pull_request_success(self, mock_post, mcp):
mock_response = MagicMock()
mock_response.status_code = 201
mock_response.json.return_value = {"number": 1, "title": "New PR"}
mock_post.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
create_pr = mcp._tool_manager._tools["github_create_pull_request"].fn
result = create_pr(
owner="owner",
repo="repo",
title="New PR",
head="feature",
base="main",
)
assert result["success"] is True
class TestGitHubSearch:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_search_code_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"total_count": 1, "items": []}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
search_code = mcp._tool_manager._tools["github_search_code"].fn
result = search_code(query="addClass")
assert result["success"] is True
class TestGitHubBranches:
@pytest.fixture
def mcp(self):
return FastMCP("test-server")
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_list_branches_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = [{"name": "main"}]
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
list_branches = mcp._tool_manager._tools["github_list_branches"].fn
result = list_branches(owner="owner", repo="repo")
assert result["success"] is True
@patch("aden_tools.tools.github_tool.github_tool.httpx.get")
def test_get_branch_success(self, mock_get, mcp):
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"name": "main", "protected": True}
mock_get.return_value = mock_response
with patch("os.getenv", return_value="ghp_test"):
register_tools(mcp, credentials=None)
get_branch = mcp._tool_manager._tools["github_get_branch"].fn
result = get_branch(owner="owner", repo="repo", branch="main")
assert result["success"] is True