Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/dataModel/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
class AccessorType(str, Enum):
OPENAI = "openai"
ANTHROPIC = "anthropic"
GITHUB_COPILOT = "github_copilot"
MOCK = "mock"

class Model(BaseModel):
Expand Down
116 changes: 116 additions & 0 deletions src/modelAccessors/github_copilot_accessor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
import json
import subprocess
from typing import Any, Optional
import os

from pydantic import TypeAdapter

from .base_accessor import BaseModelAccessor, Tool
from src.dataModel.model_response import ModelResponse


class GitHubCopilotAccessor(BaseModelAccessor):
"""GitHub Copilot accessor that creates agentic tasks and returns issue IDs."""

def __init__(self):
# Verify that gh CLI is available
try:
subprocess.run(['gh', '--version'], capture_output=True, text=True, check=True)
self._gh_available = True
except (subprocess.CalledProcessError, FileNotFoundError):
self._gh_available = False

# GitHub Copilot models that support tool usage
self.tool_supported_models = ["gpt-4", "gpt-3.5-turbo", "gpt-4o"]

def call_model(
self,
prompt: str,
*,
adapter: TypeAdapter[ModelResponse],
schema: dict,
model: str = "gpt-4",
system_prompt: str = "",
tools: Optional[list[Tool]] = None,
) -> ModelResponse:
"""
Call GitHub Copilot to create an agentic task.

This uses the gh copilot CLI to create a GitHub issue with an agentic task
that will handle the code generation. The response contains the issue ID
that was created.
"""

if not self._gh_available:
raise RuntimeError(
"GitHub CLI (gh) is not available. Please install it and authenticate with 'gh auth login'. "
"See https://github.com/cli/cli#installation for installation instructions."
)

# Combine system prompt and user prompt for the task description
full_prompt = f"{system_prompt}\n\n{prompt}".strip()

try:
# Create a GitHub Copilot agentic task
# The command format: gh copilot agent-task create "<description>" [--follow]
cmd = ['gh', 'copilot', 'agent-task', 'create', full_prompt]

# Add --follow flag to track the task progress
cmd.append('--follow')

result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True,
env=os.environ.copy()
)

output = result.stdout.strip()

# Parse the output to extract the issue ID
# The gh copilot agent-task create command should return an issue URL or ID
issue_id = self._extract_issue_id(output)

# Create a response with the issue ID
response_data = {
"type": "implemented",
"content": f"GitHub Copilot agentic task created. Issue ID: {issue_id}",
"artifacts": [issue_id]
}

return adapter.validate_python(response_data)

except subprocess.CalledProcessError as e:
# If the command fails, it might be because the extension or feature is not available
raise RuntimeError(
f"GitHub Copilot agent-task creation failed: {e.stderr}\n"
"Make sure you have the GitHub Copilot CLI extension installed: "
"gh extension install github/gh-copilot"
) from e

def _extract_issue_id(self, output: str) -> str:
"""
Extract the issue ID from gh copilot output.

The output might contain a URL like https://github.com/owner/repo/issues/123
or just the issue number.
"""
import re

# Try to find issue URL pattern
url_match = re.search(r'github\.com/[^/]+/[^/]+/issues/(\d+)', output)
if url_match:
return url_match.group(1)

# Try to find standalone issue number
num_match = re.search(r'#(\d+)', output)
if num_match:
return num_match.group(1)

# If no pattern matches, return the full output
return output

def supports_tools(self, model: str) -> bool:
"""Check if model supports tools."""
return model in self.tool_supported_models
3 changes: 3 additions & 0 deletions src/orchestrator/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from src.modelAccessors.base_accessor import BaseModelAccessor
from src.modelAccessors.openai_accessor import OpenAIAccessor
from src.modelAccessors.anthropic_accessor import AnthropicAccessor
from src.modelAccessors.github_copilot_accessor import GitHubCopilotAccessor
from src.modelAccessors.mock_accessor import MockAccessor
from src.agentNodes.clarifier import Clarifier
from src.agentNodes.hld_designer import HLDDesigner
Expand Down Expand Up @@ -126,6 +127,8 @@ def _get_accessor(self, accessor_type: AccessorType) -> BaseModelAccessor:
return OpenAIAccessor()
case AccessorType.ANTHROPIC:
return AnthropicAccessor()
case AccessorType.GITHUB_COPILOT:
return GitHubCopilotAccessor()
case AccessorType.MOCK:
return MockAccessor()
case _:
Expand Down
30 changes: 30 additions & 0 deletions src/tools/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,22 @@
read_directory,
write_directory,
)
from .github_tools import (
GitHubIssueManager,
get_issue,
create_issue,
update_issue,
comment_on_issue,
list_issues,
close_issue,
GET_ISSUE_TOOL,
CREATE_ISSUE_TOOL,
UPDATE_ISSUE_TOOL,
COMMENT_ISSUE_TOOL,
LIST_ISSUES_TOOL,
CLOSE_ISSUE_TOOL,
GITHUB_TOOLS,
)

__all__ = [
"WEB_SEARCH_TOOL",
Expand All @@ -31,4 +47,18 @@
"WRITE_DIRECTORY_TOOL",
"read_directory",
"write_directory",
"GitHubIssueManager",
"get_issue",
"create_issue",
"update_issue",
"comment_on_issue",
"list_issues",
"close_issue",
"GET_ISSUE_TOOL",
"CREATE_ISSUE_TOOL",
"UPDATE_ISSUE_TOOL",
"COMMENT_ISSUE_TOOL",
"LIST_ISSUES_TOOL",
"CLOSE_ISSUE_TOOL",
"GITHUB_TOOLS",
]
174 changes: 174 additions & 0 deletions src/tools/github_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
import json
import os
from typing import Any, Dict, Optional

import requests

from src.modelAccessors.data.tool import Tool


class GitHubIssueManager:
"""Manager for GitHub issue operations using GitHub API."""

@staticmethod
def _get_api_headers() -> Dict[str, str]:
"""Get headers for GitHub API requests."""
token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
if not token:
raise RuntimeError(
"GitHub token not found. Set GITHUB_TOKEN or GH_TOKEN environment variable. "
"You can create a token at https://github.com/settings/tokens"
)

return {
"Authorization": f"token {token}",
"Accept": "application/vnd.github.v3+json",
"Content-Type": "application/json"
}

@staticmethod
def _make_api_request(method: str, url: str, data: Optional[Dict] = None) -> Dict[str, Any]:
"""Make a request to the GitHub API."""
headers = GitHubIssueManager._get_api_headers()

try:
if method.upper() == "GET":
response = requests.get(url, headers=headers)
elif method.upper() == "POST":
response = requests.post(url, headers=headers, json=data)
elif method.upper() == "PATCH":
response = requests.patch(url, headers=headers, json=data)
else:
raise ValueError(f"Unsupported HTTP method: {method}")

response.raise_for_status()
return response.json() if response.content else {}
except requests.exceptions.RequestException as e:
raise RuntimeError(f"GitHub API request failed: {e}") from e


def get_issue(repo: str, issue_number: int) -> Dict[str, Any]:
"""Get details of a specific GitHub issue."""
url = f"https://api.github.com/repos/{repo}/issues/{issue_number}"
return GitHubIssueManager._make_api_request("GET", url)


def create_issue(repo: str, title: str, body: str = "", labels: Optional[list[str]] = None) -> Dict[str, Any]:
"""Create a new GitHub issue."""
url = f"https://api.github.com/repos/{repo}/issues"
data = {
"title": title,
"body": body
}
if labels:
data["labels"] = labels

return GitHubIssueManager._make_api_request("POST", url, data)


def update_issue(repo: str, issue_number: int, title: Optional[str] = None, body: Optional[str] = None, state: Optional[str] = None) -> Dict[str, Any]:
"""Update a GitHub issue."""
url = f"https://api.github.com/repos/{repo}/issues/{issue_number}"
data = {}

if title:
data["title"] = title
if body:
data["body"] = body
if state:
data["state"] = state

return GitHubIssueManager._make_api_request("PATCH", url, data)


def comment_on_issue(repo: str, issue_number: int, comment: str) -> Dict[str, Any]:
"""Add a comment to a GitHub issue."""
url = f"https://api.github.com/repos/{repo}/issues/{issue_number}/comments"
data = {"body": comment}
return GitHubIssueManager._make_api_request("POST", url, data)


def list_issues(repo: str, state: str = "open", limit: int = 10) -> list[Dict[str, Any]]:
"""List GitHub issues."""
url = f"https://api.github.com/repos/{repo}/issues?state={state}&per_page={limit}"
result = GitHubIssueManager._make_api_request("GET", url)
# GitHub API returns a list directly
return result if isinstance(result, list) else [result]


def close_issue(repo: str, issue_number: int) -> Dict[str, Any]:
"""Close a GitHub issue."""
return update_issue(repo, issue_number, state="closed")


# Tool definitions for use with model accessors
GET_ISSUE_TOOL = Tool(
name="get_github_issue",
description="Get details of a specific GitHub issue by repository and issue number",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"issue_number": {"type": "integer", "description": "Issue number"}
}
)

CREATE_ISSUE_TOOL = Tool(
name="create_github_issue",
description="Create a new GitHub issue",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"title": {"type": "string", "description": "Issue title"},
"body": {"type": "string", "description": "Issue body/description"},
"labels": {"type": "array", "items": {"type": "string"}, "description": "Optional labels for the issue"}
}
)

UPDATE_ISSUE_TOOL = Tool(
name="update_github_issue",
description="Update an existing GitHub issue",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"issue_number": {"type": "integer", "description": "Issue number"},
"title": {"type": "string", "description": "New title (optional)"},
"body": {"type": "string", "description": "New body/description (optional)"},
"state": {"type": "string", "enum": ["open", "closed"], "description": "New state (optional)"}
}
)

COMMENT_ISSUE_TOOL = Tool(
name="comment_github_issue",
description="Add a comment to a GitHub issue",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"issue_number": {"type": "integer", "description": "Issue number"},
"comment": {"type": "string", "description": "Comment text"}
}
)

LIST_ISSUES_TOOL = Tool(
name="list_github_issues",
description="List GitHub issues in a repository",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"state": {"type": "string", "enum": ["open", "closed", "all"], "description": "Issue state filter"},
"limit": {"type": "integer", "description": "Maximum number of issues to return"}
}
)

CLOSE_ISSUE_TOOL = Tool(
name="close_github_issue",
description="Close a GitHub issue",
parameters={
"repo": {"type": "string", "description": "Repository in format 'owner/repo'"},
"issue_number": {"type": "integer", "description": "Issue number"}
}
)

# All GitHub tools for easy import
GITHUB_TOOLS = [
GET_ISSUE_TOOL,
CREATE_ISSUE_TOOL,
UPDATE_ISSUE_TOOL,
COMMENT_ISSUE_TOOL,
LIST_ISSUES_TOOL,
CLOSE_ISSUE_TOOL,
]
Loading
Loading