Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e3c71f77de | |||
| b09824faec | |||
| 0715fc5498 | |||
| f9fddd6663 | |||
| 58b60b84fd | |||
| 86aef3319f | |||
| bfb660275e | |||
| d6ae48bc58 |
+16
-10
@@ -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"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ from framework.graph.plan import Plan
|
||||
from framework.testing.prompts import (
|
||||
PYTEST_TEST_FILE_HEADER,
|
||||
)
|
||||
from framework.utils.io import atomic_write
|
||||
|
||||
# Initialize MCP server
|
||||
mcp = FastMCP("agent-builder")
|
||||
@@ -122,11 +123,11 @@ def _save_session(session: BuildSession):
|
||||
|
||||
# Save session file
|
||||
session_file = SESSIONS_DIR / f"{session.id}.json"
|
||||
with open(session_file, "w") as f:
|
||||
with atomic_write(session_file) as f:
|
||||
json.dump(session.to_dict(), f, indent=2, default=str)
|
||||
|
||||
# Update active session pointer
|
||||
with open(ACTIVE_SESSION_FILE, "w") as f:
|
||||
with atomic_write(ACTIVE_SESSION_FILE) as f:
|
||||
f.write(session.id)
|
||||
|
||||
|
||||
@@ -246,7 +247,7 @@ def load_session_by_id(session_id: Annotated[str, "ID of the session to load"])
|
||||
_session = _load_session(session_id)
|
||||
|
||||
# Update active session pointer
|
||||
with open(ACTIVE_SESSION_FILE, "w") as f:
|
||||
with atomic_write(ACTIVE_SESSION_FILE) as f:
|
||||
f.write(session_id)
|
||||
|
||||
return json.dumps(
|
||||
@@ -1488,13 +1489,13 @@ def export_graph() -> str:
|
||||
|
||||
# Write agent.json
|
||||
agent_json_path = exports_dir / "agent.json"
|
||||
with open(agent_json_path, "w") as f:
|
||||
with atomic_write(agent_json_path) as f:
|
||||
json.dump(export_data, f, indent=2, default=str)
|
||||
|
||||
# Generate README.md
|
||||
readme_content = _generate_readme(session, export_data, all_tools)
|
||||
readme_path = exports_dir / "README.md"
|
||||
with open(readme_path, "w") as f:
|
||||
with atomic_write(readme_path) as f:
|
||||
f.write(readme_content)
|
||||
|
||||
# Write mcp_servers.json if MCP servers are configured
|
||||
@@ -1503,8 +1504,9 @@ def export_graph() -> str:
|
||||
if session.mcp_servers:
|
||||
mcp_config = {"servers": session.mcp_servers}
|
||||
mcp_servers_path = exports_dir / "mcp_servers.json"
|
||||
with open(mcp_servers_path, "w") as f:
|
||||
with atomic_write(mcp_servers_path) as f:
|
||||
json.dump(mcp_config, f, indent=2)
|
||||
|
||||
mcp_servers_size = mcp_servers_path.stat().st_size
|
||||
|
||||
# Get file sizes
|
||||
|
||||
@@ -9,6 +9,7 @@ import json
|
||||
from pathlib import Path
|
||||
|
||||
from framework.schemas.run import Run, RunStatus, RunSummary
|
||||
from framework.utils.io import atomic_write
|
||||
|
||||
|
||||
class FileStorage:
|
||||
@@ -86,13 +87,13 @@ class FileStorage:
|
||||
"""Save a run to storage."""
|
||||
# Save full run using Pydantic's model_dump_json
|
||||
run_path = self.base_path / "runs" / f"{run.id}.json"
|
||||
with open(run_path, "w", encoding="utf-8") as f:
|
||||
with atomic_write(run_path) as f:
|
||||
f.write(run.model_dump_json(indent=2))
|
||||
|
||||
# Save summary
|
||||
summary = RunSummary.from_run(run)
|
||||
summary_path = self.base_path / "summaries" / f"{run.id}.json"
|
||||
with open(summary_path, "w", encoding="utf-8") as f:
|
||||
with atomic_write(summary_path) as f:
|
||||
f.write(summary.model_dump_json(indent=2))
|
||||
|
||||
# Update indexes
|
||||
@@ -188,8 +189,8 @@ class FileStorage:
|
||||
values = self._get_index(index_type, key) # Already validated in _get_index
|
||||
if value not in values:
|
||||
values.append(value)
|
||||
with open(index_path, "w", encoding="utf-8") as f:
|
||||
json.dump(values, f)
|
||||
with atomic_write(index_path) as f:
|
||||
json.dump(values, f, indent=2)
|
||||
|
||||
def _remove_from_index(self, index_type: str, key: str, value: str) -> None:
|
||||
"""Remove a value from an index."""
|
||||
@@ -198,8 +199,8 @@ class FileStorage:
|
||||
values = self._get_index(index_type, key) # Already validated in _get_index
|
||||
if value in values:
|
||||
values.remove(value)
|
||||
with open(index_path, "w", encoding="utf-8") as f:
|
||||
json.dump(values, f)
|
||||
with atomic_write(index_path) as f:
|
||||
json.dump(values, f, indent=2)
|
||||
|
||||
# === UTILITY ===
|
||||
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@contextmanager
|
||||
def atomic_write(path: Path, mode: str = "w", encoding: str = "utf-8"):
|
||||
tmp_path = path.with_suffix(path.suffix + ".tmp")
|
||||
try:
|
||||
with open(tmp_path, mode, encoding=encoding) as f:
|
||||
yield f
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
tmp_path.replace(path)
|
||||
except BaseException:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
raise
|
||||
@@ -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=[
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -34,7 +34,7 @@ def register_tools(mcp: FastMCP) -> None:
|
||||
Returns:
|
||||
dict with success status, data, and metadata
|
||||
"""
|
||||
if (offset < 0 or (limit is not None and limit < 0)):
|
||||
if offset < 0 or (limit is not None and limit < 0):
|
||||
return {"error": "offset and limit must be non-negative"}
|
||||
try:
|
||||
secure_path = get_secure_path(path, workspace_id, agent_id, session_id)
|
||||
|
||||
@@ -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)}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user