diff --git a/.github/workflows/reindex_on_merge.yml b/.github/workflows/reindex_on_merge.yml
new file mode 100644
index 0000000..40005eb
--- /dev/null
+++ b/.github/workflows/reindex_on_merge.yml
@@ -0,0 +1,42 @@
+# .github/workflows/reindex_on_merge.yml
+
+name: Re-index Repository on Merge (Self-Hosted)
+
+on:
+ pull_request:
+ types: [closed]
+ branches:
+ - main
+ workflow_dispatch:
+
+jobs:
+ reindex:
+ # This condition ensures the job only runs if the pull request was actually merged.
+ if: github.event.pull_request.merged == true || github.event_name == 'workflow_dispatch'
+
+ # *** KEY CHANGE ***
+ # This tells GitHub to run this job on one of your self-hosted runners.
+ # You can also add labels to target specific servers, e.g., [self-hosted, linux, x64, my-app]
+ runs-on: inference-server
+
+ steps:
+ # Step 1: Check out the repository's code
+ # This downloads the latest version of your 'main' branch into the runner's working directory.
+ - name: Checkout repository
+ uses: actions/checkout@v4
+
+ # Step 2: Run the indexing script
+ # This executes your 'create_index.py' script using the Python environment on your server.
+ # It assumes Python and all dependencies from requirements.txt are already installed on the server.
+ # The GITHUB_TOKEN is still passed securely to the script.
+ - name: Run indexing script
+ run: python create_index.py
+ env:
+ GITHUB_REPOSITORY: ${{ github.repository }}
+ GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ # Optional: Specify the working directory if your bot lives in a subfolder
+ # working-directory: ./path/to/your/bot
+
+ # The "Upload database artifact" step is no longer needed, as the database
+ # is now being written directly to a persistent location on your server.
+
diff --git a/create_index.py b/create_index.py
new file mode 100644
index 0000000..d1292c7
--- /dev/null
+++ b/create_index.py
@@ -0,0 +1,185 @@
+import os
+import logging
+import torch
+import gc # Import the garbage collection module
+import chromadb
+from chromadb.utils import embedding_functions
+from dotenv import load_dotenv
+
+# Import the GitHubTool from its location
+# Assuming it's in a 'tools' directory as per your other script
+from tools.github_tool import GitHubTool
+
+# --- Configuration ---
+# You can adjust these settings
+
+# If you have downloaded a model, provide the local path here.
+# Otherwise, the model will be downloaded from Hugging Face.
+# Example: EMBEDDING_MODEL_PATH = "/path/to/your/models/all-MiniLM-L6-v2"
+EMBEDDING_MODEL_PATH = """C:\Models\embeddings\Qwen3-Embedding-0.6B"""
+
+# Path to store the local vector database
+CHROMA_DB_PATH = "C:\Models\embeddings\embedding_result\chroma_db"
+# Name of the collection within the database
+CHROMA_COLLECTION_NAME = "github_repo"
+# Files with these extensions will be indexed. Add any other text-based files you need.
+# Excludes common binary/unwanted files.
+INCLUDED_EXTENSIONS = ['.py', '.js', '.ts', '.md', '.txt', '.html', '.css', '.go', '.rs', '.java', '.c', '.h', '.cpp', '.sh', '.yaml', '.json']
+
+# *** NEW: Intelligent Chunking Configuration ***
+CHUNK_SIZE = 1000 # The target size for each text chunk in characters
+CHUNK_OVERLAP = 200 # The number of characters to overlap between chunks
+CHUNK_PROCESSING_BATCH_SIZE = 100 # The number of chunks to process in a single batch
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def get_all_repo_files(github_tool, repo_path=""):
+ """
+ Recursively fetches all file paths from the GitHub repository.
+ """
+ all_files = []
+ try:
+ items = github_tool.execute("list_files", path=repo_path)
+ if isinstance(items, str) and items.startswith("Error"):
+ logging.error(f"Could not list files at path '{repo_path}': {items}")
+ return []
+
+ for item in items:
+ # Only index files with allowed extensions
+ if item['type'] == 'file' and any(item['name'].endswith(ext) for ext in INCLUDED_EXTENSIONS):
+ all_files.append(item['path'])
+ elif item['type'] == 'dir':
+ # It's a directory, so recurse into it
+ all_files.extend(get_all_repo_files(github_tool, repo_path=item['path']))
+
+ return all_files
+ except Exception as e:
+ logging.error(f"An unexpected error occurred while listing files at '{repo_path}': {e}")
+ return all_files
+
+def split_text(text: str) -> list[str]:
+ """
+ Splits text into chunks of a specified size with overlap.
+ """
+ chunks = []
+ if text is None or not text.strip():
+ return []
+
+ start = 0
+ while start < len(text):
+ end = start + CHUNK_SIZE
+ chunks.append(text[start:end])
+ start += CHUNK_SIZE - CHUNK_OVERLAP
+ return chunks
+
+def main():
+ """
+ Main function to initialize the database and index the GitHub repository.
+ """
+ load_dotenv()
+ logging.info("Starting repository indexing process...")
+
+ # 1. Initialize GitHub Tool
+ try:
+ github_repo = os.getenv("GITHUB_REPOSITORY")
+ github_token = os.getenv("GITHUB_TOKEN")
+ if not github_repo or not github_token:
+ raise ValueError("GITHUB_REPOSITORY and GITHUB_TOKEN environment variables are required.")
+
+ github_tool = GitHubTool(repo=github_repo, token=github_token)
+ logging.info(f"Successfully initialized GitHubTool for repo: {github_repo}")
+ except Exception as e:
+ logging.fatal(f"Failed to initialize GitHubTool: {e}")
+ return
+
+ # 2. Initialize ChromaDB and Embedding Model
+ client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
+ device = "cuda" if torch.cuda.is_available() else "cpu"
+ logging.info(f"Using device: {device} for embedding model inference.")
+ model_location = EMBEDDING_MODEL_PATH
+ logging.info(f"Loading embedding model from: {model_location}")
+
+ sentence_transformer_ef = embedding_functions.SentenceTransformerEmbeddingFunction(
+ model_name=model_location,
+ device=device
+ )
+
+ logging.info(f"Loading or creating Chroma collection: '{CHROMA_COLLECTION_NAME}'")
+ collection = client.get_or_create_collection(
+ name=CHROMA_COLLECTION_NAME,
+ embedding_function=sentence_transformer_ef,
+ metadata={"hnsw:space": "cosine"}
+ )
+
+ # 3. Fetch all file paths from the repository
+ logging.info("Fetching all file paths from the repository...")
+ file_paths = get_all_repo_files(github_tool)
+ if not file_paths:
+ logging.warning("No files found to index. Exiting.")
+ return
+ logging.info(f"Found {len(file_paths)} files to potentially index.")
+
+ # 4. Process files and upsert to ChromaDB in chunk-based batches
+
+ batch_documents = []
+ batch_metadatas = []
+ batch_ids = []
+
+ for i, file_path in enumerate(file_paths):
+ logging.info(f"Processing file {i+1}/{len(file_paths)}: {file_path}")
+ try:
+ content = github_tool.execute("read_file", path=file_path)
+ if not isinstance(content, str) or content.startswith("Error"):
+ logging.warning(f"Could not read or empty content for {file_path}. Skipping.")
+ continue
+
+ # *** USE THE NEW, ROBUST CHUNKING METHOD ***
+ chunks = split_text(content)
+
+ for chunk_index, chunk in enumerate(chunks):
+ # Add the processed chunk to the current batch
+ unique_id = f"{file_path}_{chunk_index}"
+ batch_documents.append(chunk)
+ batch_metadatas.append({"source": file_path, "chunk_index": chunk_index})
+ batch_ids.append(unique_id)
+
+ # If the batch reaches the desired size, upsert it to the database
+ if len(batch_documents) >= CHUNK_PROCESSING_BATCH_SIZE:
+ logging.info(f"Upserting batch of {len(batch_documents)} chunks...")
+ collection.upsert(
+ ids=batch_ids,
+ documents=batch_documents,
+ metadatas=batch_metadatas
+ )
+ # Clear the batch lists to free up memory
+ batch_documents, batch_metadatas, batch_ids = [], [], []
+
+ # Force garbage collection and empty CUDA cache
+ logging.info("Cleaning up memory...")
+ gc.collect()
+ if device == 'cuda':
+ torch.cuda.empty_cache()
+ logging.info("Batch upserted and memory cleared.")
+
+ except Exception as e:
+ logging.error(f"Error processing file {file_path}: {e}")
+
+ # 5. Upsert any remaining documents after the loop finishes
+ if batch_documents:
+ logging.info(f"Upserting final batch of {len(batch_documents)} chunks...")
+ collection.upsert(
+ ids=batch_ids,
+ documents=batch_documents,
+ metadatas=batch_metadatas
+ )
+ # Final cleanup
+ gc.collect()
+ if device == 'cuda':
+ torch.cuda.empty_cache()
+ logging.info("Final batch upserted.")
+
+ logging.info("--- Indexing Complete ---")
+ logging.info(f"Total documents in collection: {collection.count()}")
+
+if __name__ == '__main__':
+ main()
diff --git a/rag_inference_bot.py b/rag_inference_bot.py
new file mode 100644
index 0000000..f005b8e
--- /dev/null
+++ b/rag_inference_bot.py
@@ -0,0 +1,219 @@
+import logging
+import chromadb
+from chromadb.utils import embedding_functions
+from inference_bot import InferenceBot # Correctly inherit from the ABC
+from FlagEmbedding import FlagReranker
+import argparse
+import os
+import importlib
+import torch
+from transformers import AutoTokenizer, AutoModelForCausalLM
+
+
+# --- RAG Configuration ---
+# Must match the settings in create_index.py
+EMBEDDING_MODEL_NAME = """C:\Models\embeddings\Qwen3-Embedding-0.6B"""
+CHROMA_DB_PATH = "C:\Models\embeddings\embedding_result\chroma_db"
+CHROMA_COLLECTION_NAME = "github_repo"
+
+# Using a powerful open-source reranker model
+RERANKER_MODEL_NAME = """C:\Models\embeddings\Qwen3-Reranker-0.6B"""
+
+# Number of initial results to fetch from the database before reranking
+N_RESULTS_TO_RETRIEVE = 25
+# Number of final results to keep after reranking
+N_RESULTS_TO_KEEP_AFTER_RERANK = 5
+# The minimum relevance score a result must have to be included in the final context
+RERANKER_SCORE_THRESHOLD = 0.5
+
+class RAGInferenceBot(InferenceBot):
+ def __init__(self):
+ """
+ Initializes the RAG components, including a custom implementation
+ for the Qwen3-Reranker based on its model card.
+ """
+ logging.info("Initializing RAG components...")
+ self._processing_status = {}
+ try:
+ # --- Embedding and Vector DB Initialization ---
+ self.chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
+ self.embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(
+ model_name=EMBEDDING_MODEL_NAME
+ )
+ self.collection = self.chroma_client.get_collection(
+ name=CHROMA_COLLECTION_NAME,
+ embedding_function=self.embedding_function
+ )
+ logging.info("Successfully connected to ChromaDB collection for RAG.")
+
+ # --- Custom Reranker Initialization (as per model card) ---
+ self.device = "cuda" if torch.cuda.is_available() else "cpu"
+ logging.info(f"Using device: {self.device} for Reranker model.")
+
+ self.rerank_tokenizer = AutoTokenizer.from_pretrained(RERANKER_MODEL_NAME, padding_side='left')
+ self.rerank_model = AutoModelForCausalLM.from_pretrained(RERANKER_MODEL_NAME, torch_dtype=torch.float16).to(self.device).eval()
+
+ # Manually set the padding token if it's missing
+ if self.rerank_tokenizer.pad_token is None:
+ logging.warning("Reranker tokenizer has no pad_token. Setting it to the eos_token.")
+ self.rerank_tokenizer.pad_token = self.rerank_tokenizer.eos_token
+
+ # Get token IDs for score calculation
+ self.token_false_id = self.rerank_tokenizer.convert_tokens_to_ids("no")
+ self.token_true_id = self.rerank_tokenizer.convert_tokens_to_ids("yes")
+ self.max_length = 8192
+
+ # Define and pre-encode the special prefixes and suffixes from the model card
+ prefix_text = "<|im_start|>system\nJudge whether the Document meets the requirements based on the Query and the Instruct provided. Note that the answer can only be \"yes\" or \"no\".<|im_end|>\n<|im_start|>user\n"
+ suffix_text = "<|im_end|>\n<|im_start|>assistant\n\n\n\n\n"
+ self.prefix_tokens = self.rerank_tokenizer.encode(prefix_text, add_special_tokens=False)
+ self.suffix_tokens = self.rerank_tokenizer.encode(suffix_text, add_special_tokens=False)
+
+ logging.info(f"Successfully initialized custom Reranker: {RERANKER_MODEL_NAME}")
+
+ except Exception as e:
+ logging.fatal(f"Failed to initialize RAG components: {e}", exc_info=True)
+ raise
+
+ # --- Implementation of all abstract methods from InferenceBot ---
+
+ @property
+ def processing_status(self):
+ return self._processing_status
+
+ def clear_conversation_history(self, user_id):
+ pass
+
+ async def switch_model(self):
+ return "This bot only performs RAG lookups and has no swappable models."
+
+ def set_processing_status(self, user_id, message_id):
+ self._processing_status[user_id] = {"processing": True, "message_id": message_id}
+
+ def clear_processing_status(self, user_id):
+ if user_id in self._processing_status:
+ del self._processing_status[user_id]
+
+ async def abort_processing(self, user_id):
+ if user_id in self.processing_status:
+ self.clear_processing_status(user_id)
+ return "Processing aborted."
+ else:
+ return "No active processing found to abort."
+
+ def get_bot_status(self):
+ return f"RAG Bot is active.\nEmbedding Model: {os.path.basename(EMBEDDING_MODEL_NAME)}\nReranker Model: {os.path.basename(RERANKER_MODEL_NAME)}"
+
+ async def start(self):
+ logging.info(f"{self.__class__.__name__} started.")
+
+ # --- Core RAG Logic ---
+
+ def _format_rerank_instruction(self, query, doc):
+ # Using the default instruction from the model card
+ instruction = 'Given a web search query, retrieve relevant passages that answer the query'
+ return f": {instruction}\n: {query}\n: {doc}"
+
+ @torch.no_grad()
+ def _compute_rerank_scores(self, pairs: list[list[str, str]]):
+ """
+ Custom score computation logic that follows the model card's example.
+ """
+ # Format all pairs with the required instruction format
+ formatted_pairs = [self._format_rerank_instruction(query, doc) for query, doc in pairs]
+
+ # Tokenize the formatted pairs
+ inputs = self.rerank_tokenizer(
+ formatted_pairs, padding=False, truncation='longest_first',
+ return_attention_mask=False, max_length=self.max_length - len(self.prefix_tokens) - len(self.suffix_tokens)
+ )
+
+ # Add special prefix and suffix tokens to each item in the batch
+ for i in range(len(inputs['input_ids'])):
+ inputs['input_ids'][i] = self.prefix_tokens + inputs['input_ids'][i] + self.suffix_tokens
+
+ # Pad the batch to the same length
+ inputs = self.rerank_tokenizer.pad(inputs, padding=True, return_tensors="pt", max_length=self.max_length)
+ inputs = {key: inputs[key].to(self.device) for key in inputs}
+
+ # Get model outputs (logits)
+ batch_scores = self.rerank_model(**inputs).logits[:, -1, :]
+
+ # Calculate scores based on the probability of "yes" vs "no"
+ true_vector = batch_scores[:, self.token_true_id]
+ false_vector = batch_scores[:, self.token_false_id]
+
+ scores = torch.stack([false_vector, true_vector], dim=1)
+ scores = torch.nn.functional.log_softmax(scores, dim=1)
+
+ final_scores = scores[:, 1].exp().tolist()
+ return final_scores
+
+ def _retrieve_and_rerank_context(self, query: str):
+ logging.info(f"RAG: Retrieving context for query: '{query}'")
+ if not query: return ""
+ try:
+ results = self.collection.query(query_texts=[query], n_results=N_RESULTS_TO_RETRIEVE)
+ initial_docs = results.get('documents', [[]])[0]
+ if not initial_docs:
+ logging.info("RAG: No initial documents found in vector search.")
+ return "No relevant context found in the knowledge base."
+
+ logging.info(f"RAG: Retrieved {len(initial_docs)} initial documents from ChromaDB.")
+
+ # Create pairs of [query, document] for reranking
+ rerank_pairs = [[query, doc] for doc in initial_docs]
+
+ # Use our custom scoring function
+ scores = self._compute_rerank_scores(rerank_pairs)
+
+ scored_docs = sorted(zip(scores, initial_docs, results['metadatas'][0]), key=lambda x: x[0], reverse=True)
+
+ # Take the top N results
+ top_k_docs = scored_docs[:N_RESULTS_TO_KEEP_AFTER_RERANK]
+
+ # *** NEW: Filter the top N results by the score threshold ***
+ final_docs = [doc for doc in top_k_docs if doc[0] >= RERANKER_SCORE_THRESHOLD]
+
+ logging.info(f"RAG: Reranked and filtered down to {len(final_docs)} documents with score >= {RERANKER_SCORE_THRESHOLD}.")
+
+ # If no documents meet the threshold, inform the user.
+ if not final_docs:
+ return "No highly relevant context found after filtering."
+
+ context_lines = []
+ for i, (score, doc, metadata) in enumerate(final_docs):
+ source = metadata.get('source', 'Unknown file')
+ chunk_index = metadata.get('chunk_index', 'N/A')
+ context_lines.append(f"--- Context {i+1} (Relevance: {score:.2f}) ---\nSource: {source}\n\n{doc}\n")
+
+ return "\n".join(context_lines)
+ except Exception as e:
+ logging.error(f"RAG: Error during context retrieval/reranking: {e}", exc_info=True)
+ return "An error occurred while searching the knowledge base."
+
+ async def handle_message(self, user_id, user_message):
+ context_response = self._retrieve_and_rerank_context(user_message)
+ return context_response
+
+
+def main_rag():
+ logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+
+ parser = argparse.ArgumentParser(description='RAG-Only Inference Bot')
+ parser.add_argument('--messenger', type=str, help='Messenger type (i.e. telegram)', required=True)
+ args = parser.parse_args()
+
+ try:
+ bot = RAGInferenceBot()
+
+ full_code_file = importlib.import_module(f'{args.messenger.lower()}_helper')
+ helper_class = getattr(full_code_file, f"{args.messenger.capitalize()}Helper")
+ helper = helper_class(bot)
+ helper.run()
+
+ except Exception as e:
+ logging.fatal(f"An unexpected error occurred during bot initialization: {e}", exc_info=True)
+
+if __name__ == '__main__':
+ main_rag()
diff --git a/requirements.txt b/requirements.txt
index d76bbcd..6836b11 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -10,3 +10,9 @@ pytest-cov
google-genai
httpx==0.27.2
tiktoken
+chromadb
+sentence-transformers
+transformers>=4.38
+torch --index-url https://download.pytorch.org/whl/cu121
+torchvision --index-url https://download.pytorch.org/whl/cu121
+torchaudio --index-url https://download.pytorch.org/whl/cu121
\ No newline at end of file
diff --git a/tools/github_tool.py b/tools/github_tool.py
index 034147c..8b6913c 100644
--- a/tools/github_tool.py
+++ b/tools/github_tool.py
@@ -1,4 +1,1275 @@
-# tools/github_tool.py\nfrom .base_tool import BaseTool\nimport requests\nimport os\nimport base64\nimport logging\n\nclass GitHubTool(BaseTool):\n def __init__(self, session=None, token=None, repo=None, base_url=None, initial_branch=\"main\", logger=None):\n self.base_url = base_url if base_url else \"https://api.github.com\"\n self._token = token if token else os.environ.get(\"GITHUB_TOKEN\")\n self._repo = repo if repo else os.environ.get(\"GITHUB_REPOSITORY\")\n \n if not self._token:\n # In a real scenario, might raise an error or operate in a degraded mode.\n # For this tool, token is essential.\n raise ValueError(\"GitHub token must be provided either as an argument or via GITHUB_TOKEN env var.\")\n if not self._repo:\n raise ValueError(\"GitHub repository (e.g., \'owner/repo\') must be provided either as an argument or via GITHUB_REPOSITORY env var.\")\n\n if session:\n self.session = session\n else:\n self.session = requests.Session()\n self.session.headers.update({\n \"Authorization\": f\"token {self._token}\",\n \"Accept\": \"application/vnd.github.v3+json\"\n })\n \n self.current_branch = initial_branch\n\n # Use provided logger or get a new one for the module\n # The application using this tool should configure the logging handlers and formatting.\n self.logger = logger if logger else logging.getLogger(__name__)\n # If no handlers are configured by the application, add a NullHandler\n # to prevent \"No handler found\" warnings if the tool logs something.\n if not self.logger.handlers:\n self.logger.addHandler(logging.NullHandler())\n\n def clear(self):\n if self.current_branch != \"main\":\n self._set_current_branch(\"main\")\n self.logger.info(f\"GitHubTool state cleared. Current branch is {self.current_branch}\")\n\n def get_functions(self):\n return [\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"read_file\",\n \"description\": \"Read a file from the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"path\": {\"type\": \"string\", \"description\": \"Path to the file in the repository\"}\n },\n \"required\": [\"path\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"read_readme\",\n \"description\": \"Read the README.md file from the root of the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {}\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"list_files\",\n \"description\": \"List files in a directory of the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"path\": {\"type\": \"string\", \"description\": \"Path to the directory in the repository\"}\n },\n \"required\": [\"path\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"search_code\",\n \"description\": \"Search for code in the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"query\": {\"type\": \"string\", \"description\": \"Search query\"}\n },\n \"required\": [\"query\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_branch\",\n \"description\": \"Create a new branch in the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"branch_name\": {\"type\": \"string\", \"description\": \"Name of the new branch\"},\n \"base_branch\": {\"type\": \"string\", \"description\": \"Name of the base branch\", \"default\": \"main\"}\n },\n \"required\": [\"branch_name\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"commit_file\",\n \"description\": \"Commit a file to a branch (not main)\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"file_path\": {\"type\": \"string\", \"description\": \"Path to the file in the repository\"},\n \"commit_message\": {\"type\": \"string\", \"description\": \"Commit message\"},\n \"content\": {\"type\": \"string\", \"description\": \"Content of the file\"}\n },\n \"required\": [\"file_path\", \"commit_message\", \"content\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_pull_request\",\n \"description\": \"Create a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"title\": {\"type\": \"string\", \"description\": \"Title of the pull request\"},\n \"body\": {\"type\": \"string\", \"description\": \"Body of the pull request\"},\n \"base\": {\"type\": \"string\", \"description\": \"The name of the branch you want the changes pulled into\", \"default\": \"main\"}\n },\
- \"required\": [\"title\", \"body\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_commit_history\",\n \"description\": \"Get commit history for a file\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"file_path\": {\"type\": \"string\", \"description\": \"Path to the file in the repository\"},\n \"num_commits\": {\"type\": \"integer\", \"description\": \"Number of commits to retrieve\", \"default\": 10}\n },\n \"required\": [\"file_path\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"view_commit_details_for_file\",\n \"description\": \"View commit history and details for a specific file, including commit messages.\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"file_path\": {\"type\": \"string\", \"description\": \"Path to the file in the repository\"},\n \"num_commits\": {\"type\": \"integer\", \"description\": \"Number of commits to retrieve\", \"default\": 10}\n },\n \"required\": [\"file_path\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_branch_sha\",\n \"description\": \"Get the SHA of the latest commit on a branch\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"branch\": {\"type\": \"string\", \"description\": \"Name of the branch\"}\n },\n \"required\": [\"branch\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_current_branch\",\n \"description\": \"Get the name of the current branch\",\n \"parameters\": { \"type\": \"object\", \"properties\": {} }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"set_current_branch\",\n \"description\": \"Set the current branch\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"branch_name\": {\"type\": \"string\", \"description\": \"Name of the branch to set as current\"}\n },\n \"required\": [\"branch_name\"]\n }\n },\n \"_tags\": [\"read\", \"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_file_at_commit\",\n \"description\": \"Get the contents of a file at a specific commit\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"file_path\": {\"type\": \"string\", \"description\": \"Path to the file in the repository\"},\n \"commit_sha\": {\"type\": \"string\", \"description\": \"SHA of the commit to retrieve the file from\"}\n },\n \"required\": [\"file_path\", \"commit_sha\"]\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"list_branches\",\n \"description\": \"List all branches in the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"per_page\": {\"type\": \"integer\", \"description\": \"Number of branches to return per page (max 100)\", \"default\": 100},\n \"all_pages\": {\"type\": \"boolean\", \"description\": \"Whether to fetch all pages of results\", \"default\": True}\n }\n }\n },\n \"_tags\": [\"read\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"approve_pull_request\",\n \"description\": \"Approve a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"close_pull_request\",\n \"description\": \"Close a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"merge_pull_request\",\n \"description\": \"Merge a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"},\n \"commit_title\": {\"type\": \"string\", \"description\": \"Title for the automatic commit message\", \"default\": \"Merge pull request\"},\n \"commit_message\": {\"type\": \"string\", \"description\": \"Extra detail to append to automatic commit message\", \"default\": \"\"},\n \"merge_method\": {\n \"type\": \"string\",\n \"description\": \"Merge method to use\",\n \"enum\": [\"merge\", \"squash\", \"rebase\"],\n \"default\": \"merge\"\n }\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"delete_branch\",\n \"description\": \"Delete a branch\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"branch_name\": {\"type\": \"string\", \"description\": \"Name of the branch to delete\"}\n },\
- \"required\": [\"branch_name\"]\n }\n },\n \"_tags\": [\"write\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_issue_details\",\n \"description\": \"Get details of a specific issue\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"issue_number\": {\"type\": \"integer\", \"description\": \"The number of the issue\"}\n },\n \"required\": [\"issue_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_issue\",\n \"description\": \"Create a new issue in the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"title\": {\"type\": \"string\", \"description\": \"Title of the issue\"},\n \"body\": {\"type\": \"string\", \"description\": \"Body of the issue\"},\n \"labels\": {\n \"type\": \"array\",\n \"items\": {\"type\": \"string\"},\n \"description\": \"Labels to apply to the issue\"\n }\n },\n \"required\": [\"title\", \"body\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"list_issues\",\n \"description\": \"List issues in the repository\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"state\": {\"type\": \"string\", \"enum\": [\"open\", \"closed\", \"all\"], \"default\": \"open\", \"description\": \"State of the issues to retrieve\"},\n \"per_page\": {\"type\": \"integer\", \"default\": 30, \"description\": \"Number of issues to return per page\"},\n \"page\": {\"type\": \"integer\", \"default\": 1, \"description\": \"Page number of the results to fetch\"}\n }\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"add_issue_comment\",\n \"description\": \"Add a comment to an issue\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"issue_number\": {\"type\": \"integer\", \"description\": \"The number of the issue\"},\n \"comment\": {\"type\": \"string\", \"description\": \"The comment to add to the issue\"}\n },\
- \"required\": [\"issue_number\", \"comment\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_issue_comments\",\n \"description\": \"Get comments for an issue\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"issue_number\": {\"type\": \"integer\", \"description\": \"The number of the issue\"}\n },\n \"required\": [\"issue_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n { \n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_pull_request_general_comments\",\n \"description\": \"Get general comments posted on a pull request itself (not specific to file lines).\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request.\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_project_board\",\n \"description\": \"Create a new project board\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"name\": {\"type\": \"string\", \"description\": \"Name of the project board\"},\n \"body\": {\"type\": \"string\", \"description\": \"Body of the project board\"}\n },\n \"required\": [\"name\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_project_column\",\n \"description\": \"Create a new column in a project board\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"project_id\": {\"type\": \"integer\", \"description\": \"ID of the project board\"},\n \"column_name\": {\"type\": \"string\", \"description\": \"Name of the column\"}\n },\n \"required\": [\"project_id\", \"column_name\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_project_card\",\n \"description\": \"Create a new card in a project column\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"column_id\": {\"type\": \"integer\", \"description\": \"ID of the project column\"},\n \"note\": {\"type\": \"string\", \"description\": \"Note for the project card\"}\n },\n \"required\": [\"column_id\", \"note\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"move_project_card\",\n \"description\": \"Move a card to a new position\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"card_id\": {\"type\": \"integer\", \"description\": \"ID of the project card\"},\n \"position\": {\"type\": \"string\", \"description\": \"New position of the card\"},\n \"column_id\": {\"type\": \"integer\", \"description\": \"ID of the target column\"}\n },\n \"required\": [\"card_id\", \"position\", \"column_id\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"link_issue_to_project_card\",\n \"description\": \"Link an issue or pull request to a project card\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"card_id\": {\"type\": \"integer\", \"description\": \"ID of the project card\"},\n \"content_id\": {\"type\": \"integer\", \"description\": \"ID of the issue or pull request\"},\n \"content_type\": {\"type\": \"string\", \"description\": \"Type of the content (Issue or PullRequest)\"}\n },\n \"required\": [\"card_id\", \"content_id\", \"content_type\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"list_project_boards\",\n \"description\": \"List project boards associated with the repository\",\n \"parameters\": { \"type\": \"object\", \"properties\": {} }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"view_project_board_items\",\n \"description\": \"View items (columns and cards) in a specific project board\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"project_id\": {\"type\": \"integer\", \"description\": \"ID of the project board\"}\n },\n \"required\": [\"project_id\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_pull_request_details\",\n \"description\": \"Get detailed information about a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_pull_request_diff\",\n \"description\": \"Get the diff of a pull request\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"get_pull_request_files\",\n \"description\": \"Get a list of files changed in a pull request, with their patch details\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"create_pull_request_review_comment\",\n \"description\": \"Add a comment to a specific line of a file in a pull request review\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"},\n \"body\": {\"type\": \"string\", \"description\": \"The text of the comment\"},\n \"commit_id\": {\"type\": \"string\", \"description\": \"The SHA of the commit to comment on\"},\n \"path\": {\"type\": \"string\", \"description\": \"The path to the file being commented on\"},\n \"position\": {\"type\": \"integer\", \"description\": \"The line index in the diff to comment on (starting at 1)\"},\n \"side\": {\"type\": \"string\", \"enum\": [\"LEFT\", \"RIGHT\"], \"description\": \"The side of the diff to comment on (LEFT for old file, RIGHT for new file)\", \"default\": \"RIGHT\"},\n \"start_line\": {\"type\": \"integer\", \"description\": \"The start line of the diff if commenting on a range\"},\n \"start_side\": {\"type\": \"string\", \"enum\": [\"LEFT\", \"RIGHT\"], \"description\": \"The side of the diff for the start line\"}\n },\n \"required\": [\"pull_number\", \"body\", \"commit_id\", \"path\", \"position\"]\n }\n },\n \"_tags\": [\"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"list_pull_request_review_comments\",\n \"description\": \"List comments on a pull request review\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"}\n },\n \"required\": [\"pull_number\"]\n }\n },\n \"_tags\": [\"read\", \"communicate\"]\n },\n {\n \"type\": \"function\",\n \"function\": {\n \"name\": \"submit_pull_request_review\",\n \"description\": \"Submit a formal pull request review (APPROVE, REQUEST_CHANGES, COMMENT)\",\n \"parameters\": {\n \"type\": \"object\",\n \"properties\": {\n \"pull_number\": {\"type\": \"integer\", \"description\": \"The number of the pull request\"},\n \"event\": {\"type\": \"string\", \"enum\": [\"APPROVE\", \"REQUEST_CHANGES\", \"COMMENT\"], \"description\": \"The type of review event\"},\n \"body\": {\"type\": \"string\", \"description\": \"The body of the review (required for REQUEST_CHANGES, optional for others)\"}\n },\n \"required\": [\"pull_number\", \"event\"]\n }\n },\n \"_tags\": [\"communicate\"]\n }\n ]\n\n def execute(self, function_name, **kwargs):\n self.logger.info(f\"Executing GitHub Tool function: {function_name} with args: {kwargs}\")\n # Dispatch to the appropriate private method\n method_name = f\"_{function_name}\"\n if hasattr(self, method_name):\n method = getattr(self, method_name)\n try:\n return method(**kwargs) # Ensure only expected args are passed if method signature is strict\n except Exception as e:\n self.logger.error(f\"Error executing {method_name}: {e}\", exc_info=True)\n return f\"Error during {function_name} execution: {str(e)}\"\n else:\n error_message = f\"Unknown function: {function_name}\"\n self.logger.error(error_message)\n return error_message\n\n # Private methods for each function, using self.session for HTTP requests\n\n def _read_file(self, path):\n self.logger.info(f\"Reading file: {path} from branch: {self.current_branch}\")\n url = f\"{self.base_url}/repos/{self._repo}/contents/{path}\"\n response = self.session.get(url, params={\"ref\": self.current_branch})\n if response.status_code == 200:\n content = response.json()[\"content\"]\n decoded_content = base64.b64decode(content).decode(\'utf-8\')\n self.logger.info(f\"Successfully read file: {path}\")\n return decoded_content\n else:\n error_message = f\"Error reading file ({path}): {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _read_readme(self):\n self.logger.info(\"Reading README.md from the root of the repository.\")\n return self._read_file(\"README.md\")\n\n def _create_branch(self, branch_name, base_branch=\"main\"):\n self.logger.info(f\"Creating branch: {branch_name} from base: {base_branch}\")\n # Get SHA of base branch\n ref_url = f\"{self.base_url}/repos/{self._repo}/git/refs/heads/{base_branch}\"\n response_sha = self.session.get(ref_url)\n if response_sha.status_code != 200:\n error_message = f\"Error getting base branch SHA ({base_branch}): {response_sha.status_code} - {response_sha.text}\"\n self.logger.error(error_message)\n return error_message\n sha = response_sha.json()[\"object\"][\"sha\"]\n \n # Create new branch\n create_ref_url = f\"{self.base_url}/repos/{self._repo}/git/refs\"\n data = {\"ref\": f\"refs/heads/{branch_name}\", \"sha\": sha}\n response_create = self.session.post(create_ref_url, json=data)\n if response_create.status_code == 201:\n self.current_branch = branch_name\n success_message = f\"Branch \'{branch_name}\' created successfully from \'{base_branch}\' and set as current branch.\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error creating branch \'{branch_name}\': {response_create.status_code} - {response_create.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _commit_file(self, file_path, content, commit_message):\n self.logger.info(f\"Committing file: {file_path} to branch: {self.current_branch} with message: \'{commit_message}\'\")\n if self.current_branch == \"main\":\n error_message = \"Action directly to main branch is not allowed. Please create and switch to a new branch first.\"\n self.logger.warning(error_message)\n return error_message\n\n url = f\"{self.base_url}/repos/{self._repo}/contents/{file_path}\"\n encoded_content = base64.b64encode(content.encode(\'utf-8\')).decode(\'utf-8\')\n data = {\n \"message\": commit_message,\n \"content\": encoded_content,\n \"branch\": self.current_branch\n }\n\n # Check if file exists to get its SHA for update\n self.logger.info(f\"Checking if file \'{file_path}\' exists on branch \'{self.current_branch}\'\")\n get_response = self.session.get(url, params={\"ref\": self.current_branch})\n if get_response.status_code == 200:\n data[\"sha\"] = get_response.json()[\"sha\"]\n self.logger.info(f\"File \'{file_path}\' exists, will update.\")\n elif get_response.status_code == 404:\n self.logger.info(f\"File \'{file_path}\' does not exist, will create.\")\n else:\n error_message = f\"Error checking file existence for \'{file_path}\': {get_response.status_code} - {get_response.text}\"\n self.logger.error(error_message)\n return error_message\n\n response = self.session.put(url, json=data)\n if response.status_code in [200, 201]: # 200 for update, 201 for create\n commit_sha = response.json().get(\"commit\", {}).get(\"sha\", \"N/A\")\n success_message = f\"File \'{file_path}\' committed successfully to branch \'{self.current_branch}\'. Commit SHA: {commit_sha}\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error committing file \'{file_path}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _create_pull_request(self, title, body, base=\"main\"):\n self.logger.info(f\"Creating pull request: \'{title}\' from branch \'{self.current_branch}\' to \'{base}\'\")\n if self.current_branch == base:\n error_message = f\"Cannot create a pull request from branch \'{self.current_branch}\' to itself (\'{base}\').\"\n self.logger.warning(error_message)\n return error_message\n \n url = f\"{self.base_url}/repos/{self._repo}/pulls\"\n data = {\"title\": title, \"body\": body, \"head\": self.current_branch, \"base\": base}\n response = self.session.post(url, json=data)\n if response.status_code == 201:\n pr_html_url = response.json().get(\"html_url\", \"N/A\")\n pr_number = response.json().get(\"number\", \"N/A\")\n success_message = f\"Pull request \'{title}\' created successfully: {pr_html_url} (Number: {pr_number})\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error creating pull request: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_branch_sha(self, branch):\n self.logger.info(f\"Getting SHA for branch: {branch}\")\n url = f\"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch}\"\n response = self.session.get(url)\n if response.status_code == 200:\n sha = response.json()[\"object\"][\"sha\"]\n self.logger.info(f\"SHA for branch \'{branch}\' is {sha}\")\n return sha\n else:\n error_message = f\"Error getting SHA for branch \'{branch}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _list_files(self, path):\n self.logger.info(f\"Listing files in path: \'{path}\' on branch: \'{self.current_branch}\'\")\n url = f\"{self.base_url}/repos/{self._repo}/contents/{path.strip(\'/\')}\" # Ensure no leading/trailing slashes for consistency\n response = self.session.get(url, params={\"ref\": self.current_branch})\n if response.status_code == 200:\n items = response.json()\n results = []\n if isinstance(items, list): # It\'s a directory listing\n for item in items:\n results.append({\"name\": item[\"name\"], \"type\": item[\"type\"], \"path\": item[\"path\"]})\n elif isinstance(items, dict) and \'type\' in items: # It\'s a single file response\n results.append({\"name\": items[\"name\"], \"type\": items[\"type\"], \"path\": items[\"path\"]})\n self.logger.info(f\"Successfully listed {len(results)} items in \'{path}\'.\")\n return results\n elif response.status_code == 404:\n self.logger.warning(f\"Path \'{path}\' not found on branch \'{self.current_branch}\'.\")\n return f\"Error: Path \'{path}\' not found.\"\n else:\n error_message = f\"Error listing files in \'{path}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _search_code(self, query):\n self.logger.info(f\"Searching code with query: \'{query}\' in repo: \'{self._repo}\'\")\n url = f\"{self.base_url}/search/code\"\n params = {\"q\": f\"{query} repo:{self._repo}\"}\n response = self.session.get(url, params=params)\n if response.status_code == 200:\n search_results = response.json().get(\"items\", [])\n results = [{\"path\": item[\"path\"], \"url\": item[\"html_url\"]} for item in search_results]\n self.logger.info(f\"Code search for \'{query}\' found {len(results)} items.\")\n return results\n else:\n error_message = f\"Error searching code for \'{query}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_commit_history(self, file_path, num_commits=10):\n self.logger.info(f\"Getting last {num_commits} commit(s) for file: \'{file_path}\' on branch \'{self.current_branch}\'\")\n url = f\"{self.base_url}/repos/{self._repo}/commits\"\n params = {\"path\": file_path, \"sha\": self.current_branch, \"per_page\": num_commits}\n response = self.session.get(url, params=params)\n if response.status_code == 200:\n commits_data = response.json()\n commits = [{\n \"sha\": commit[\"sha\"],\n \"message\": commit[\"commit\"][\"message\"],\n \"author\": commit[\"commit\"][\"author\"][\"name\"],\n \"date\": commit[\"commit\"][\"author\"][\"date\"]\n } for commit in commits_data]\n self.logger.info(f\"Successfully retrieved {len(commits)} commit(s) for \'{file_path}\'.\")\n return commits\n else:\n error_message = f\"Error getting commit history for \'{file_path}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _view_commit_details_for_file(self, file_path, num_commits=10):\n # This function is essentially the same as get_commit_history based on its description.\n self.logger.info(f\"Viewing commit details for file \'{file_path}\' (last {num_commits} commits) - using _get_commit_history.\")\n return self._get_commit_history(file_path, num_commits)\n\n def _get_current_branch(self):\n self.logger.info(f\"Current branch is: {self.current_branch}\")\n return self.current_branch\n\n def _set_current_branch(self, branch_name):\n self.logger.info(f\"Attempting to set current branch to: {branch_name}\")\n # Check if branch exists by trying to get its SHA\n sha_info = self._get_branch_sha(branch_name)\n if isinstance(sha_info, str) and sha_info.startswith(\"Error getting SHA\"): # Crude check for error string\n error_message = f\"Cannot set current branch: Branch \'{branch_name}\' not found or error accessing it. Details: {sha_info}\"\n self.logger.warning(error_message)\n return error_message \n \n self.current_branch = branch_name\n success_message = f\"Current branch set to: {self.current_branch}\"\n self.logger.info(success_message)\n return success_message\n\n def _get_file_at_commit(self, file_path, commit_sha):\n self.logger.info(f\"Getting file \'{file_path}\' at commit SHA: {commit_sha}\")\n url = f\"{self.base_url}/repos/{self._repo}/contents/{file_path}\"\n response = self.session.get(url, params={\"ref\": commit_sha})\n if response.status_code == 200:\n content = response.json()[\"content\"]\n decoded_content = base64.b64decode(content).decode(\'utf-8\')\n self.logger.info(f\"Successfully retrieved file \'{file_path}\' at commit {commit_sha}.\")\n return decoded_content\n else:\n error_message = f\"Error reading file \'{file_path}\' at commit {commit_sha}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _list_branches(self, per_page=100, all_pages=True):\n self.logger.info(f\"Listing branches for repo \'{self._repo}\'. Per_page={per_page}, All_pages={all_pages}\")\n url = f\"{self.base_url}/repos/{self._repo}/branches\"\n params = {\"per_page\": min(per_page, 100)} # Respect GitHub API limit\n branches_list = []\n page = 1\n while url:\n self.logger.debug(f\"Fetching page {page} from {url} with params {params if page==1 else {}}\")\n response = self.session.get(url, params=params if page == 1 else None) # params only for first page if paginating via links\n if response.status_code != 200:\n error_message = f\"Error listing branches: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n \n current_page_branches = [branch[\"name\"] for branch in response.json()]\n branches_list.extend(current_page_branches)\n self.logger.debug(f\"Fetched {len(current_page_branches)} branches on page {page}.\")\n\n if not all_pages or not response.links.get(\"next\"):\n break\n url = response.links[\"next\"][\"url\"]\n page += 1\n params = {} # Clear params for subsequent calls using a link that includes them\n\n self.logger.info(f\"Successfully listed {len(branches_list)} branches.\")\n return branches_list\n\n def _approve_pull_request(self, pull_number):\n self.logger.info(f\"Approving pull request #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews\"\n data = {\"event\": \"APPROVE\"}\n response = self.session.post(url, json=data)\n if response.status_code == 200:\n success_message = f\"Pull request #{pull_number} approved successfully.\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error approving pull request #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _close_pull_request(self, pull_number):\n self.logger.info(f\"Closing pull request #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}\"\n data = {\"state\": \"closed\"}\n response = self.session.patch(url, json=data) # Use PATCH for update\n if response.status_code == 200:\n success_message = f\"Pull request #{pull_number} closed successfully.\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error closing pull request #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _merge_pull_request(self, pull_number, commit_title=\"Merge pull request\", commit_message=\"\", merge_method=\"merge\"):\n self.logger.info(f\"Merging pull request #{pull_number} using method \'{merge_method}\'\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/merge\"\n data = {\"commit_title\": commit_title, \"commit_message\": commit_message, \"merge_method\": merge_method}\n response = self.session.put(url, json=data)\n if response.status_code == 200:\n success_message = f\"Pull request #{pull_number} merged successfully.\"\n self.logger.info(success_message)\n return success_message\n elif response.status_code == 405: # Method Not Allowed (e.g., PR not mergeable)\n error_message = f\"Error merging pull request #{pull_number}: Not mergeable. {response.json().get(\'message\', response.text)}\"\n self.logger.warning(error_message)\n return error_message\n elif response.status_code == 409: # Conflict\n error_message = f\"Error merging pull request #{pull_number}: Merge conflict. {response.json().get(\'message\', response.text)}\"\n self.logger.warning(error_message)\n return error_message\n else:\n error_message = f\"Error merging pull request #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _delete_branch(self, branch_name):\n self.logger.info(f\"Deleting branch: {branch_name}\")\n if branch_name == \"main\" or (hasattr(self, \'default_branch\') and branch_name == self.default_branch) :\n # Add a check for a configurable default branch if necessary\n error_message = f\"Cannot delete protected branch: {branch_name}\"\n self.logger.warning(error_message)\n return error_message\n\n url = f\"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch_name}\"\n response = self.session.delete(url)\n if response.status_code == 204:\n success_message = f\"Branch \'{branch_name}\' deleted successfully.\"\n self.logger.info(success_message)\n if self.current_branch == branch_name:\n self.current_branch = \"main\" # Or some other default\n self.logger.info(f\"Current branch was {branch_name}, reset to {self.current_branch}.\")\n return success_message\n else:\n error_message = f\"Error deleting branch \'{branch_name}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_issue_details(self, issue_number):\n self.logger.info(f\"Getting details for issue #{issue_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/issues/{issue_number}\"\n response = self.session.get(url)\n if response.status_code == 200:\n self.logger.info(f\"Successfully retrieved details for issue #{issue_number}.\")\n return response.json() # Return raw JSON data for now\n else:\n error_message = f\"Error getting details for issue #{issue_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _create_issue(self, title, body, labels=None):\n self.logger.info(f\"Creating new issue with title: \'{title}\'\")\n url = f\"{self.base_url}/repos/{self._repo}/issues\"\n data = {\"title\": title, \"body\": body}\n if labels: # Ensure labels is a list of strings\n data[\"labels\"] = labels if isinstance(labels, list) else [labels]\n \n response = self.session.post(url, json=data)\n if response.status_code == 201:\n issue_html_url = response.json().get(\"html_url\", \"N/A\")\n issue_number = response.json().get(\"number\", \"N/A\")\n success_message = f\"Issue \'{title}\' created successfully: {issue_html_url} (Number: {issue_number})\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error creating issue \'{title}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _list_issues(self, state=\"open\", per_page=30, page=1):\n self.logger.info(f\"Listing issues with state: {state}, per_page: {per_page}, page: {page}\")\n url = f\"{self.base_url}/repos/{self._repo}/issues\"\n params = {\"state\": state, \"per_page\": per_page, \"page\": page}\n response = self.session.get(url, params=params)\n if response.status_code == 200:\n issues_data = response.json()\n self.logger.info(f\"Successfully listed {len(issues_data)} issues.\")\n # Return a summary or full data based on needs\n return [{ \"title\": i[\"title\"], \"number\": i[\"number\"], \"state\": i[\"state\"], \"url\": i[\"html_url\"] } for i in issues_data]\n else:\n error_message = f\"Error listing issues: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _add_issue_comment(self, issue_number, comment):\n self.logger.info(f\"Adding comment to issue #{issue_number}: \'{comment[:50]}...\'\")\n url = f\"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments\"\n data = {\"body\": comment}\n response = self.session.post(url, json=data)\n if response.status_code == 201:\n comment_html_url = response.json().get(\"html_url\", \"N/A\")\n success_message = f\"Comment added to issue #{issue_number} successfully: {comment_html_url}\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error adding comment to issue #{issue_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_issue_comments(self, issue_number):\n self.logger.info(f\"Getting comments for issue #{issue_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments\"\n response = self.session.get(url)\n if response.status_code == 200:\n comments_data = response.json()\n self.logger.info(f\"Successfully retrieved {len(comments_data)} comments for issue #{issue_number}.\")\n # Return summary or full data\n return [{ \"user\": c[\"user\"][\"login\"], \"body\": c[\"body\"], \"created_at\": c[\"created_at\"] } for c in comments_data]\n else:\n error_message = f\"Error getting comments for issue #{issue_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_pull_request_general_comments(self, pull_number):\n self.logger.info(f\"Getting general comments for pull request #{pull_number}\")\n # In GitHub API, PR comments (general, not review comments on lines) are issue comments.\n # The PR is also an issue, so use the issue comments endpoint.\n return self._get_issue_comments(issue_number=pull_number)\n\n def _create_project_board(self, name, body=None):\n self.logger.info(f\"Creating project board: \'{name}\'\")\n url = f\"{self.base_url}/repos/{self._repo}/projects\"\n headers = self.session.headers.copy() # Get existing session headers\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\" # Required for Projects API\n data = {\"name\": name}\n if body: data[\"body\"] = body\n response = self.session.post(url, headers=headers, json=data)\n if response.status_code == 201:\n project_data = response.json()\n success_message = f\"Project board \'{name}\' created successfully with ID: {project_data[\'id\']}\"\n self.logger.info(success_message)\n return project_data # Return full project data\n else:\n error_message = f\"Error creating project board \'{name}\': {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _create_project_column(self, project_id, column_name):\n self.logger.info(f\"Creating column \'{column_name}\' for project ID: {project_id}\")\n url = f\"{self.base_url}/projects/{project_id}/columns\"\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n data = {\"name\": column_name}\n response = self.session.post(url, headers=headers, json=data)\n if response.status_code == 201:\n column_data = response.json()\n success_message = f\"Column \'{column_name}\' created successfully for project {project_id} with ID: {column_data[\'id\']}\"\n self.logger.info(success_message)\n return column_data\n else:\n error_message = f\"Error creating column \'{column_name}\' for project {project_id}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _create_project_card(self, column_id, note=None, content_id=None, content_type=None):\n self.logger.info(f\"Creating card in column ID: {column_id}\")\n url = f\"{self.base_url}/projects/columns/{column_id}/cards\"\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n data = {}\n if note:\n data[\"note\"] = note\n if content_id and content_type:\n data[\"content_id\"] = content_id\n data[\"content_type\"] = content_type\n elif (content_id and not content_type) or (not content_id and content_type):\n err = \"Both content_id and content_type must be provided to link content to a project card.\"\n self.logger.warning(err)\n return err\n \n if not data:\n return \"Error: Card must have a note or content to link.\"\n\n response = self.session.post(url, headers=headers, json=data)\n if response.status_code == 201:\n card_data = response.json()\n success_message = f\"Card created successfully in column {column_id} with ID: {card_data[\'id\']}\"\n self.logger.info(success_message)\n return card_data\n else:\n error_message = f\"Error creating card in column {column_id}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _move_project_card(self, card_id, position, column_id=None):\n self.logger.info(f\"Moving card ID: {card_id} to position: {position}\" + (f\" in column ID: {column_id}\" if column_id else \"\"))\n url = f\"{self.base_url}/projects/columns/cards/{card_id}/moves\"\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n data = {\"position\": position}\n if column_id:\n data[\"column_id\"] = column_id\n \n response = self.session.post(url, headers=headers, json=data)\n if response.status_code == 201: # Successful move returns 201 with empty body\n success_message = f\"Card {card_id} moved successfully to position {position}\" + (f\" in column {column_id}\" if column_id else \".\")\n self.logger.info(success_message)\n return success_message # Return success message as body is empty\n else:\n error_message = f\"Error moving card {card_id}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n # _link_issue_to_project_card is effectively handled by _create_project_card if content_id and content_type are passed.\n # The API used to have a separate link endpoint, but now it is part of card creation/update.\n # For updating an existing card to link an issue, one would PATCH the card\'s content_id/content_type.\n # Let\'s assume the function intends to update an existing card if it\'s a separate function.\n # However, the provided API spec for `link_issue_to_project_card` uses PATCH on card_id, so let\'s implement that.\n def _link_issue_to_project_card(self, card_id, content_id, content_type):\n self.logger.info(f\"Linking content_id {content_id} (type: {content_type}) to card_id {card_id}\")\n url = f\"{self.base_url}/projects/cards/{card_id}\" # Note: API docs suggest /projects/columns/cards/{card_id} or /projects/cards/{card_id}\n # Using /projects/cards/{card_id} as it seems more general for card update.\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n data = {\"content_id\": content_id, \"content_type\": content_type}\n\n response = self.session.patch(url, headers=headers, json=data)\n if response.status_code == 200:\n updated_card = response.json()\n success_message = f\"{content_type} {content_id} linked to card {card_id} successfully.\"\n self.logger.info(success_message)\n return updated_card\n else:\n error_message = f\"Error linking {content_type} {content_id} to card {card_id}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _list_project_boards(self):\n self.logger.info(f\"Listing project boards for repo: {self._repo}\")\n url = f\"{self.base_url}/repos/{self._repo}/projects\"\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n response = self.session.get(url, headers=headers)\n if response.status_code == 200:\n projects_data = response.json()\n self.logger.info(f\"Successfully listed {len(projects_data)} project boards.\")\n return projects_data\n else:\n error_message = f\"Error listing project boards: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _view_project_board_items(self, project_id):\n self.logger.info(f\"Viewing items for project ID: {project_id}\")\n columns_url = f\"{self.base_url}/projects/{project_id}/columns\"\n headers = self.session.headers.copy()\n headers[\"Accept\"] = \"application/vnd.github.inertia-preview+json\"\n \n columns_response = self.session.get(columns_url, headers=headers)\n if columns_response.status_code != 200:\n error_message = f\"Error fetching columns for project {project_id}: {columns_response.status_code} - {columns_response.text}\"\n self.logger.error(error_message)\n return error_message\n \n columns_data = columns_response.json()\n project_items = []\n for column in columns_data:\n column_info = {\"id\": column[\"id\"], \"name\": column[\"name\"], \"cards\": []}\n cards_url = column[\"cards_url\"]\n cards_response = self.session.get(cards_url, headers=headers)\n if cards_response.status_code == 200:\n column_info[\"cards\"] = cards_response.json()\n else:\n self.logger.error(f\"Error fetching cards for column {column[\'id\']}(\'{column[\'name\']}\'): {cards_response.status_code} - {cards_response.text}\")\n column_info[\"cards\"] = \"Error fetching cards\"\n project_items.append(column_info)\n \n self.logger.info(f\"Successfully retrieved items for project ID: {project_id}.\")\n return project_items\n\n def _get_pull_request_details(self, pull_number):\n self.logger.info(f\"Getting details for PR #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}\"\n response = self.session.get(url)\n if response.status_code == 200:\n self.logger.info(f\"Successfully retrieved details for PR #{pull_number}.\")\n return response.json()\n else:\n error_message = f\"Error getting details for PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_pull_request_diff(self, pull_number):\n self.logger.info(f\"Getting diff for PR #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}\"\n diff_headers = self.session.headers.copy()\n diff_headers[\"Accept\"] = \"application/vnd.github.diff\"\n response = self.session.get(url, headers=diff_headers)\n if response.status_code == 200:\n self.logger.info(f\"Successfully retrieved diff for PR #{pull_number}.\")\n return response.text\n else:\n error_message = f\"Error getting diff for PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _get_pull_request_files(self, pull_number):\n self.logger.info(f\"Getting files for PR #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/files\"\n response = self.session.get(url)\n if response.status_code == 200:\n self.logger.info(f\"Successfully retrieved files for PR #{pull_number}.\")\n return response.json()\n else:\n error_message = f\"Error getting files for PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _create_pull_request_review_comment(self, pull_number, body, commit_id, path, position, side=\"RIGHT\", start_line=None, start_side=None):\n self.logger.info(f\"Creating review comment on PR #{pull_number}, file \'{path}\', position {position}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments\"\n data = {\"body\": body, \"commit_id\": commit_id, \"path\": path, \"position\": position, \"side\": side}\n if start_line is not None: data[\"start_line\"] = start_line\n if start_side is not None: data[\"start_side\"] = start_side\n \n response = self.session.post(url, json=data)\n if response.status_code == 201:\n comment_url = response.json().get(\"html_url\", \"N/A\")\n success_message = f\"Review comment created successfully on PR #{pull_number}: {comment_url}\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error creating review comment on PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _list_pull_request_review_comments(self, pull_number):\n self.logger.info(f\"Listing review comments for PR #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments\"\n response = self.session.get(url)\n if response.status_code == 200:\n self.logger.info(f\"Successfully retrieved review comments for PR #{pull_number}.\")\n return response.json()\n else:\n error_message = f\"Error listing review comments for PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n\n def _submit_pull_request_review(self, pull_number, event, body=None):\n self.logger.info(f\"Submitting \'{event}\' review for PR #{pull_number}\")\n url = f\"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews\"\n data = {\"event\": event.upper()} # Ensure event is uppercase as per API\n if body: data[\"body\"] = body\n \n response = self.session.post(url, json=data)\n if response.status_code == 200:\n review_url = response.json().get(\"html_url\", \"N/A\")\n success_message = f\"Review ({event}) submitted successfully for PR #{pull_number}: {review_url}\"\n self.logger.info(success_message)\n return success_message\n else:\n error_message = f\"Error submitting review for PR #{pull_number}: {response.status_code} - {response.text}\"\n self.logger.error(error_message)\n return error_message\n
\ No newline at end of file
+# tools/github_tool.py
+from .base_tool import BaseTool
+import requests
+import os
+import base64
+import logging
+
+class GitHubTool(BaseTool):
+ def __init__(self, session=None, token=None, repo=None, base_url=None, initial_branch="main", logger=None):
+ self.base_url = base_url if base_url else "https://api.github.com"
+ self._token = token if token else os.environ.get("GITHUB_TOKEN")
+ self._repo = repo if repo else os.environ.get("GITHUB_REPOSITORY")
+
+ if not self._token:
+ # In a real scenario, might raise an error or operate in a degraded mode.
+ # For this tool, token is essential.
+ raise ValueError("GitHub token must be provided either as an argument or via GITHUB_TOKEN env var.")
+ if not self._repo:
+ raise ValueError("GitHub repository (e.g., 'owner/repo') must be provided either as an argument or via GITHUB_REPOSITORY env var.")
+
+ if session:
+ self.session = session
+ else:
+ self.session = requests.Session()
+ self.session.headers.update({
+ "Authorization": f"token {self._token}",
+ "Accept": "application/vnd.github.v3+json"
+ })
+
+ self.current_branch = initial_branch
+
+ # Use provided logger or get a new one for the module
+ # The application using this tool should configure the logging handlers and formatting.
+ self.logger = logger if logger else logging.getLogger(__name__)
+ # If no handlers are configured by the application, add a NullHandler
+ # to prevent "No handler found" warnings if the tool logs something.
+ if not self.logger.handlers:
+ self.logger.addHandler(logging.NullHandler())
+
+ def clear(self):
+ if self.current_branch != "main":
+ self._set_current_branch("main")
+ self.logger.info(f"GitHubTool state cleared. Current branch is {self.current_branch}")
+
+ def get_functions(self):
+ return [
+ {
+ "type": "function",
+ "function": {
+ "name": "read_file",
+ "description": "Read a file from the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "path": {"type": "string", "description": "Path to the file in the repository"}
+ },
+ "required": ["path"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "read_readme",
+ "description": "Read the README.md file from the root of the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {}
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "list_files",
+ "description": "List files in a directory of the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "path": {"type": "string", "description": "Path to the directory in the repository"}
+ },
+ "required": ["path"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "search_code",
+ "description": "Search for code in the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "query": {"type": "string", "description": "Search query"}
+ },
+ "required": ["query"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_branch",
+ "description": "Create a new branch in the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "branch_name": {"type": "string", "description": "Name of the new branch"},
+ "base_branch": {"type": "string", "description": "Name of the base branch", "default": "main"}
+ },
+ "required": ["branch_name"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "commit_file",
+ "description": "Commit a file to a branch (not main)",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "file_path": {"type": "string", "description": "Path to the file in the repository"},
+ "commit_message": {"type": "string", "description": "Commit message"},
+ "content": {"type": "string", "description": "Content of the file"}
+ },
+ "required": ["file_path", "commit_message", "content"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_pull_request",
+ "description": "Create a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "title": {"type": "string", "description": "Title of the pull request"},
+ "body": {"type": "string", "description": "Body of the pull request"},
+ "base": {"type": "string", "description": "The name of the branch you want the changes pulled into", "default": "main"}
+ },
+ "required": ["title", "body"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_commit_history",
+ "description": "Get commit history for a file",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "file_path": {"type": "string", "description": "Path to the file in the repository"},
+ "num_commits": {"type": "integer", "description": "Number of commits to retrieve", "default": 10}
+ },
+ "required": ["file_path"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "view_commit_details_for_file",
+ "description": "View commit history and details for a specific file, including commit messages.",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "file_path": {"type": "string", "description": "Path to the file in the repository"},
+ "num_commits": {"type": "integer", "description": "Number of commits to retrieve", "default": 10}
+ },
+ "required": ["file_path"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_branch_sha",
+ "description": "Get the SHA of the latest commit on a branch",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "branch": {"type": "string", "description": "Name of the branch"}
+ },
+ "required": ["branch"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_current_branch",
+ "description": "Get the name of the current branch",
+ "parameters": {"type": "object", "properties": {}}
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "set_current_branch",
+ "description": "Set the current branch",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "branch_name": {"type": "string", "description": "Name of the branch to set as current"}
+ },
+ "required": ["branch_name"]
+ }
+ },
+ "_tags": ["read", "write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_file_at_commit",
+ "description": "Get the contents of a file at a specific commit",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "file_path": {"type": "string", "description": "Path to the file in the repository"},
+ "commit_sha": {"type": "string", "description": "SHA of the commit to retrieve the file from"}
+ },
+ "required": ["file_path", "commit_sha"]
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "list_branches",
+ "description": "List all branches in the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "per_page": {"type": "integer", "description": "Number of branches to return per page (max 100)", "default": 100},
+ "all_pages": {"type": "boolean", "description": "Whether to fetch all pages of results", "default": True}
+ }
+ }
+ },
+ "_tags": ["read"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "approve_pull_request",
+ "description": "Approve a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "close_pull_request",
+ "description": "Close a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "merge_pull_request",
+ "description": "Merge a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"},
+ "commit_title": {"type": "string", "description": "Title for the automatic commit message", "default": "Merge pull request"},
+ "commit_message": {"type": "string", "description": "Extra detail to append to automatic commit message", "default": ""},
+ "merge_method": {
+ "type": "string",
+ "description": "Merge method to use",
+ "enum": ["merge", "squash", "rebase"],
+ "default": "merge"
+ }
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "delete_branch",
+ "description": "Delete a branch",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "branch_name": {"type": "string", "description": "Name of the branch to delete"}
+ },
+ "required": ["branch_name"]
+ }
+ },
+ "_tags": ["write"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_issue_details",
+ "description": "Get details of a specific issue",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "issue_number": {"type": "integer", "description": "The number of the issue"}
+ },
+ "required": ["issue_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_issue",
+ "description": "Create a new issue in the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "title": {"type": "string", "description": "Title of the issue"},
+ "body": {"type": "string", "description": "Body of the issue"},
+ "labels": {
+ "type": "array",
+ "items": {"type": "string"},
+ "description": "Labels to apply to the issue"
+ }
+ },
+ "required": ["title", "body"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "list_issues",
+ "description": "List issues in the repository",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "state": {"type": "string", "enum": ["open", "closed", "all"], "default": "open", "description": "State of the issues to retrieve"},
+ "per_page": {"type": "integer", "default": 30, "description": "Number of issues to return per page"},
+ "page": {"type": "integer", "default": 1, "description": "Page number of the results to fetch"}
+ }
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "add_issue_comment",
+ "description": "Add a comment to an issue",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "issue_number": {"type": "integer", "description": "The number of the issue"},
+ "comment": {"type": "string", "description": "The comment to add to the issue"}
+ },
+ "required": ["issue_number", "comment"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_issue_comments",
+ "description": "Get comments for an issue",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "issue_number": {"type": "integer", "description": "The number of the issue"}
+ },
+ "required": ["issue_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_pull_request_general_comments",
+ "description": "Get general comments posted on a pull request itself (not specific to file lines).",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request."}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_project_board",
+ "description": "Create a new project board",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "name": {"type": "string", "description": "Name of the project board"},
+ "body": {"type": "string", "description": "Body of the project board"}
+ },
+ "required": ["name"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_project_column",
+ "description": "Create a new column in a project board",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "project_id": {"type": "integer", "description": "ID of the project board"},
+ "column_name": {"type": "string", "description": "Name of the column"}
+ },
+ "required": ["project_id", "column_name"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_project_card",
+ "description": "Create a new card in a project column",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "column_id": {"type": "integer", "description": "ID of the project column"},
+ "note": {"type": "string", "description": "Note for the project card"}
+ },
+ "required": ["column_id", "note"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "move_project_card",
+ "description": "Move a card to a new position",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "card_id": {"type": "integer", "description": "ID of the project card"},
+ "position": {"type": "string", "description": "New position of the card"},
+ "column_id": {"type": "integer", "description": "ID of the target column"}
+ },
+ "required": ["card_id", "position", "column_id"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "link_issue_to_project_card",
+ "description": "Link an issue or pull request to a project card",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "card_id": {"type": "integer", "description": "ID of the project card"},
+ "content_id": {"type": "integer", "description": "ID of the issue or pull request"},
+ "content_type": {"type": "string", "description": "Type of the content (Issue or PullRequest)"}
+ },
+ "required": ["card_id", "content_id", "content_type"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "list_project_boards",
+ "description": "List project boards associated with the repository",
+ "parameters": {"type": "object", "properties": {}}
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "view_project_board_items",
+ "description": "View items (columns and cards) in a specific project board",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "project_id": {"type": "integer", "description": "ID of the project board"}
+ },
+ "required": ["project_id"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_pull_request_details",
+ "description": "Get detailed information about a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_pull_request_diff",
+ "description": "Get the diff of a pull request",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "get_pull_request_files",
+ "description": "Get a list of files changed in a pull request, with their patch details",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "create_pull_request_review_comment",
+ "description": "Add a comment to a specific line of a file in a pull request review",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"},
+ "body": {"type": "string", "description": "The text of the comment"},
+ "commit_id": {"type": "string", "description": "The SHA of the commit to comment on"},
+ "path": {"type": "string", "description": "The path to the file being commented on"},
+ "position": {"type": "integer", "description": "The line index in the diff to comment on (starting at 1)"},
+ "side": {"type": "string", "enum": ["LEFT", "RIGHT"], "description": "The side of the diff to comment on (LEFT for old file, RIGHT for new file)", "default": "RIGHT"},
+ "start_line": {"type": "integer", "description": "The start line of the diff if commenting on a range"},
+ "start_side": {"type": "string", "enum": ["LEFT", "RIGHT"], "description": "The side of the diff for the start line"}
+ },
+ "required": ["pull_number", "body", "commit_id", "path", "position"]
+ }
+ },
+ "_tags": ["communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "list_pull_request_review_comments",
+ "description": "List comments on a pull request review",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"}
+ },
+ "required": ["pull_number"]
+ }
+ },
+ "_tags": ["read", "communicate"]
+ },
+ {
+ "type": "function",
+ "function": {
+ "name": "submit_pull_request_review",
+ "description": "Submit a formal pull request review (APPROVE, REQUEST_CHANGES, COMMENT)",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "pull_number": {"type": "integer", "description": "The number of the pull request"},
+ "event": {"type": "string", "enum": ["APPROVE", "REQUEST_CHANGES", "COMMENT"], "description": "The type of review event"},
+ "body": {"type": "string", "description": "The body of the review (required for REQUEST_CHANGES, optional for others)"}
+ },
+ "required": ["pull_number", "event"]
+ }
+ },
+ "_tags": ["communicate"]
+ }
+ ]
+
+ def execute(self, function_name, **kwargs):
+ self.logger.info(f"Executing GitHub Tool function: {function_name} with args: {kwargs}")
+ # Dispatch to the appropriate private method
+ method_name = f"_{function_name}"
+ if hasattr(self, method_name):
+ method = getattr(self, method_name)
+ try:
+ return method(**kwargs) # Ensure only expected args are passed if method signature is strict
+ except Exception as e:
+ self.logger.error(f"Error executing {method_name}: {e}", exc_info=True)
+ return f"Error during {function_name} execution: {str(e)}"
+ else:
+ error_message = f"Unknown function: {function_name}"
+ self.logger.error(error_message)
+ return error_message
+
+ # Private methods for each function, using self.session for HTTP requests
+
+ def _read_file(self, path):
+ self.logger.info(f"Reading file: {path} from branch: {self.current_branch}")
+ url = f"{self.base_url}/repos/{self._repo}/contents/{path}"
+ response = self.session.get(url, params={"ref": self.current_branch})
+ if response.status_code == 200:
+ content = response.json()["content"]
+ decoded_content = base64.b64decode(content).decode('utf-8')
+ self.logger.info(f"Successfully read file: {path}")
+ return decoded_content
+ else:
+ error_message = f"Error reading file ({path}): {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _read_readme(self):
+ self.logger.info("Reading README.md from the root of the repository.")
+ return self._read_file("README.md")
+
+ def _create_branch(self, branch_name, base_branch="main"):
+ self.logger.info(f"Creating branch: {branch_name} from base: {base_branch}")
+ # Get SHA of base branch
+ ref_url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{base_branch}"
+ response_sha = self.session.get(ref_url)
+ if response_sha.status_code != 200:
+ error_message = f"Error getting base branch SHA ({base_branch}): {response_sha.status_code} - {response_sha.text}"
+ self.logger.error(error_message)
+ return error_message
+ sha = response_sha.json()["object"]["sha"]
+
+ # Create new branch
+ create_ref_url = f"{self.base_url}/repos/{self._repo}/git/refs"
+ data = {"ref": f"refs/heads/{branch_name}", "sha": sha}
+ response_create = self.session.post(create_ref_url, json=data)
+ if response_create.status_code == 201:
+ self.current_branch = branch_name
+ success_message = f"Branch '{branch_name}' created successfully from '{base_branch}' and set as current branch."
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error creating branch '{branch_name}': {response_create.status_code} - {response_create.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _commit_file(self, file_path, content, commit_message):
+ self.logger.info(f"Committing file: {file_path} to branch: {self.current_branch} with message: '{commit_message}'")
+ if self.current_branch == "main":
+ error_message = "Action directly to main branch is not allowed. Please create and switch to a new branch first."
+ self.logger.warning(error_message)
+ return error_message
+
+ url = f"{self.base_url}/repos/{self._repo}/contents/{file_path}"
+ encoded_content = base64.b64encode(content.encode('utf-8')).decode('utf-8')
+ data = {
+ "message": commit_message,
+ "content": encoded_content,
+ "branch": self.current_branch
+ }
+
+ # Check if file exists to get its SHA for update
+ self.logger.info(f"Checking if file '{file_path}' exists on branch '{self.current_branch}'")
+ get_response = self.session.get(url, params={"ref": self.current_branch})
+ if get_response.status_code == 200:
+ data["sha"] = get_response.json()["sha"]
+ self.logger.info(f"File '{file_path}' exists, will update.")
+ elif get_response.status_code == 404:
+ self.logger.info(f"File '{file_path}' does not exist, will create.")
+ else:
+ error_message = f"Error checking file existence for '{file_path}': {get_response.status_code} - {get_response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ response = self.session.put(url, json=data)
+ if response.status_code in [200, 201]: # 200 for update, 201 for create
+ commit_sha = response.json().get("commit", {}).get("sha", "N/A")
+ success_message = f"File '{file_path}' committed successfully to branch '{self.current_branch}'. Commit SHA: {commit_sha}"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error committing file '{file_path}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _create_pull_request(self, title, body, base="main"):
+ self.logger.info(f"Creating pull request: '{title}' from branch '{self.current_branch}' to '{base}'")
+ if self.current_branch == base:
+ error_message = f"Cannot create a pull request from branch '{self.current_branch}' to itself ('{base}')."
+ self.logger.warning(error_message)
+ return error_message
+
+ url = f"{self.base_url}/repos/{self._repo}/pulls"
+ data = {"title": title, "body": body, "head": self.current_branch, "base": base}
+ response = self.session.post(url, json=data)
+ if response.status_code == 201:
+ pr_html_url = response.json().get("html_url", "N/A")
+ pr_number = response.json().get("number", "N/A")
+ success_message = f"Pull request '{title}' created successfully: {pr_html_url} (Number: {pr_number})"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error creating pull request: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_branch_sha(self, branch):
+ self.logger.info(f"Getting SHA for branch: {branch}")
+ url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch}"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ sha = response.json()["object"]["sha"]
+ self.logger.info(f"SHA for branch '{branch}' is {sha}")
+ return sha
+ else:
+ error_message = f"Error getting SHA for branch '{branch}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _list_files(self, path):
+ self.logger.info(f"Listing files in path: '{path}' on branch: '{self.current_branch}'")
+ url = f"{self.base_url}/repos/{self._repo}/contents/{path.strip('/')}" # Ensure no leading/trailing slashes for consistency
+ response = self.session.get(url, params={"ref": self.current_branch})
+ if response.status_code == 200:
+ items = response.json()
+ results = []
+ if isinstance(items, list): # It's a directory listing
+ for item in items:
+ results.append({"name": item["name"], "type": item["type"], "path": item["path"]})
+ elif isinstance(items, dict) and 'type' in items: # It's a single file response
+ results.append({"name": items["name"], "type": items["type"], "path": items["path"]})
+ self.logger.info(f"Successfully listed {len(results)} items in '{path}'.")
+ return results
+ elif response.status_code == 404:
+ self.logger.warning(f"Path '{path}' not found on branch '{self.current_branch}'.")
+ return f"Error: Path '{path}' not found."
+ else:
+ error_message = f"Error listing files in '{path}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _search_code(self, query):
+ self.logger.info(f"Searching code with query: '{query}' in repo: '{self._repo}'")
+ url = f"{self.base_url}/search/code"
+ params = {"q": f"{query} repo:{self._repo}"}
+ response = self.session.get(url, params=params)
+ if response.status_code == 200:
+ search_results = response.json().get("items", [])
+ results = [{"path": item["path"], "url": item["html_url"]} for item in search_results]
+ self.logger.info(f"Code search for '{query}' found {len(results)} items.")
+ return results
+ else:
+ error_message = f"Error searching code for '{query}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_commit_history(self, file_path, num_commits=10):
+ self.logger.info(f"Getting last {num_commits} commit(s) for file: '{file_path}' on branch '{self.current_branch}'")
+ url = f"{self.base_url}/repos/{self._repo}/commits"
+ params = {"path": file_path, "sha": self.current_branch, "per_page": num_commits}
+ response = self.session.get(url, params=params)
+ if response.status_code == 200:
+ commits_data = response.json()
+ commits = [{
+ "sha": commit["sha"],
+ "message": commit["commit"]["message"],
+ "author": commit["commit"]["author"]["name"],
+ "date": commit["commit"]["author"]["date"]
+ } for commit in commits_data]
+ self.logger.info(f"Successfully retrieved {len(commits)} commit(s) for '{file_path}'.")
+ return commits
+ else:
+ error_message = f"Error getting commit history for '{file_path}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _view_commit_details_for_file(self, file_path, num_commits=10):
+ # This function is essentially the same as get_commit_history based on its description.
+ self.logger.info(f"Viewing commit details for file '{file_path}' (last {num_commits} commits) - using _get_commit_history.")
+ return self._get_commit_history(file_path, num_commits)
+
+ def _get_current_branch(self):
+ self.logger.info(f"Current branch is: {self.current_branch}")
+ return self.current_branch
+
+ def _set_current_branch(self, branch_name):
+ self.logger.info(f"Attempting to set current branch to: {branch_name}")
+ # Check if branch exists by trying to get its SHA
+ sha_info = self._get_branch_sha(branch_name)
+ if isinstance(sha_info, str) and sha_info.startswith("Error getting SHA"): # Crude check for error string
+ error_message = f"Cannot set current branch: Branch '{branch_name}' not found or error accessing it. Details: {sha_info}"
+ self.logger.warning(error_message)
+ return error_message
+
+ self.current_branch = branch_name
+ success_message = f"Current branch set to: {self.current_branch}"
+ self.logger.info(success_message)
+ return success_message
+
+ def _get_file_at_commit(self, file_path, commit_sha):
+ self.logger.info(f"Getting file '{file_path}' at commit SHA: {commit_sha}")
+ url = f"{self.base_url}/repos/{self._repo}/contents/{file_path}"
+ response = self.session.get(url, params={"ref": commit_sha})
+ if response.status_code == 200:
+ content = response.json()["content"]
+ decoded_content = base64.b64decode(content).decode('utf-8')
+ self.logger.info(f"Successfully retrieved file '{file_path}' at commit {commit_sha}.")
+ return decoded_content
+ else:
+ error_message = f"Error reading file '{file_path}' at commit {commit_sha}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _list_branches(self, per_page=100, all_pages=True):
+ self.logger.info(f"Listing branches for repo '{self._repo}'. Per_page={per_page}, All_pages={all_pages}")
+ url = f"{self.base_url}/repos/{self._repo}/branches"
+ params = {"per_page": min(per_page, 100)} # Respect GitHub API limit
+ branches_list = []
+ page = 1
+ while url:
+ self.logger.debug(f"Fetching page {page} from {url} with params {params if page==1 else {}}")
+ response = self.session.get(url, params=params if page == 1 else None) # params only for first page if paginating via links
+ if response.status_code != 200:
+ error_message = f"Error listing branches: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ current_page_branches = [branch["name"] for branch in response.json()]
+ branches_list.extend(current_page_branches)
+ self.logger.debug(f"Fetched {len(current_page_branches)} branches on page {page}.")
+
+ if not all_pages or not response.links.get("next"):
+ break
+ url = response.links["next"]["url"]
+ page += 1
+ params = {} # Clear params for subsequent calls using a link that includes them
+
+ self.logger.info(f"Successfully listed {len(branches_list)} branches.")
+ return branches_list
+
+ def _approve_pull_request(self, pull_number):
+ self.logger.info(f"Approving pull request #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews"
+ data = {"event": "APPROVE"}
+ response = self.session.post(url, json=data)
+ if response.status_code == 200:
+ success_message = f"Pull request #{pull_number} approved successfully."
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error approving pull request #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _close_pull_request(self, pull_number):
+ self.logger.info(f"Closing pull request #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}"
+ data = {"state": "closed"}
+ response = self.session.patch(url, json=data) # Use PATCH for update
+ if response.status_code == 200:
+ success_message = f"Pull request #{pull_number} closed successfully."
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error closing pull request #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _merge_pull_request(self, pull_number, commit_title="Merge pull request", commit_message="", merge_method="merge"):
+ self.logger.info(f"Merging pull request #{pull_number} using method '{merge_method}'")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/merge"
+ data = {"commit_title": commit_title, "commit_message": commit_message, "merge_method": merge_method}
+ response = self.session.put(url, json=data)
+ if response.status_code == 200:
+ success_message = f"Pull request #{pull_number} merged successfully."
+ self.logger.info(success_message)
+ return success_message
+ elif response.status_code == 405: # Method Not Allowed (e.g., PR not mergeable)
+ error_message = f"Error merging pull request #{pull_number}: Not mergeable. {response.json().get('message', response.text)}"
+ self.logger.warning(error_message)
+ return error_message
+ elif response.status_code == 409: # Conflict
+ error_message = f"Error merging pull request #{pull_number}: Merge conflict. {response.json().get('message', response.text)}"
+ self.logger.warning(error_message)
+ return error_message
+ else:
+ error_message = f"Error merging pull request #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _delete_branch(self, branch_name):
+ self.logger.info(f"Deleting branch: {branch_name}")
+ if branch_name == "main" or (hasattr(self, 'default_branch') and branch_name == self.default_branch):
+ # Add a check for a configurable default branch if necessary
+ error_message = f"Cannot delete protected branch: {branch_name}"
+ self.logger.warning(error_message)
+ return error_message
+
+ url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch_name}"
+ response = self.session.delete(url)
+ if response.status_code == 204:
+ success_message = f"Branch '{branch_name}' deleted successfully."
+ self.logger.info(success_message)
+ if self.current_branch == branch_name:
+ self.current_branch = "main" # Or some other default
+ self.logger.info(f"Current branch was {branch_name}, reset to {self.current_branch}.")
+ return success_message
+ else:
+ error_message = f"Error deleting branch '{branch_name}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_issue_details(self, issue_number):
+ self.logger.info(f"Getting details for issue #{issue_number}")
+ url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ self.logger.info(f"Successfully retrieved details for issue #{issue_number}.")
+ return response.json() # Return raw JSON data for now
+ else:
+ error_message = f"Error getting details for issue #{issue_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _create_issue(self, title, body, labels=None):
+ self.logger.info(f"Creating new issue with title: '{title}'")
+ url = f"{self.base_url}/repos/{self._repo}/issues"
+ data = {"title": title, "body": body}
+ if labels: # Ensure labels is a list of strings
+ data["labels"] = labels if isinstance(labels, list) else [labels]
+
+ response = self.session.post(url, json=data)
+ if response.status_code == 201:
+ issue_html_url = response.json().get("html_url", "N/A")
+ issue_number = response.json().get("number", "N/A")
+ success_message = f"Issue '{title}' created successfully: {issue_html_url} (Number: {issue_number})"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error creating issue '{title}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _list_issues(self, state="open", per_page=30, page=1):
+ self.logger.info(f"Listing issues with state: {state}, per_page: {per_page}, page: {page}")
+ url = f"{self.base_url}/repos/{self._repo}/issues"
+ params = {"state": state, "per_page": per_page, "page": page}
+ response = self.session.get(url, params=params)
+ if response.status_code == 200:
+ issues_data = response.json()
+ self.logger.info(f"Successfully listed {len(issues_data)} issues.")
+ # Return a summary or full data based on needs
+ return [{"title": i["title"], "number": i["number"], "state": i["state"], "url": i["html_url"]} for i in issues_data]
+ else:
+ error_message = f"Error listing issues: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _add_issue_comment(self, issue_number, comment):
+ self.logger.info(f"Adding comment to issue #{issue_number}: '{comment[:50]}...'")
+ url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments"
+ data = {"body": comment}
+ response = self.session.post(url, json=data)
+ if response.status_code == 201:
+ comment_html_url = response.json().get("html_url", "N/A")
+ success_message = f"Comment added to issue #{issue_number} successfully: {comment_html_url}"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error adding comment to issue #{issue_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_issue_comments(self, issue_number):
+ self.logger.info(f"Getting comments for issue #{issue_number}")
+ url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ comments_data = response.json()
+ self.logger.info(f"Successfully retrieved {len(comments_data)} comments for issue #{issue_number}.")
+ # Return summary or full data
+ return [{"user": c["user"]["login"], "body": c["body"], "created_at": c["created_at"]} for c in comments_data]
+ else:
+ error_message = f"Error getting comments for issue #{issue_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_pull_request_general_comments(self, pull_number):
+ self.logger.info(f"Getting general comments for pull request #{pull_number}")
+ # In GitHub API, PR comments (general, not review comments on lines) are issue comments.
+ # The PR is also an issue, so use the issue comments endpoint.
+ return self._get_issue_comments(issue_number=pull_number)
+
+ def _create_project_board(self, name, body=None):
+ self.logger.info(f"Creating project board: '{name}'")
+ url = f"{self.base_url}/repos/{self._repo}/projects"
+ headers = self.session.headers.copy() # Get existing session headers
+ headers["Accept"] = "application/vnd.github.inertia-preview+json" # Required for Projects API
+ data = {"name": name}
+ if body:
+ data["body"] = body
+ response = self.session.post(url, headers=headers, json=data)
+ if response.status_code == 201:
+ project_data = response.json()
+ success_message = f"Project board '{name}' created successfully with ID: {project_data['id']}"
+ self.logger.info(success_message)
+ return project_data # Return full project data
+ else:
+ error_message = f"Error creating project board '{name}': {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _create_project_column(self, project_id, column_name):
+ self.logger.info(f"Creating column '{column_name}' for project ID: {project_id}")
+ url = f"{self.base_url}/projects/{project_id}/columns"
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+ data = {"name": column_name}
+ response = self.session.post(url, headers=headers, json=data)
+ if response.status_code == 201:
+ column_data = response.json()
+ success_message = f"Column '{column_name}' created successfully for project {project_id} with ID: {column_data['id']}"
+ self.logger.info(success_message)
+ return column_data
+ else:
+ error_message = f"Error creating column '{column_name}' for project {project_id}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _create_project_card(self, column_id, note=None, content_id=None, content_type=None):
+ self.logger.info(f"Creating card in column ID: {column_id}")
+ url = f"{self.base_url}/projects/columns/{column_id}/cards"
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+ data = {}
+ if note:
+ data["note"] = note
+ if content_id and content_type:
+ data["content_id"] = content_id
+ data["content_type"] = content_type
+ elif (content_id and not content_type) or (not content_id and content_type):
+ err = "Both content_id and content_type must be provided to link content to a project card."
+ self.logger.warning(err)
+ return err
+
+ if not data:
+ return "Error: Card must have a note or content to link."
+
+ response = self.session.post(url, headers=headers, json=data)
+ if response.status_code == 201:
+ card_data = response.json()
+ success_message = f"Card created successfully in column {column_id} with ID: {card_data['id']}"
+ self.logger.info(success_message)
+ return card_data
+ else:
+ error_message = f"Error creating card in column {column_id}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _move_project_card(self, card_id, position, column_id=None):
+ self.logger.info(f"Moving card ID: {card_id} to position: {position}" + (f" in column ID: {column_id}" if column_id else ""))
+ url = f"{self.base_url}/projects/columns/cards/{card_id}/moves"
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+ data = {"position": position}
+ if column_id:
+ data["column_id"] = column_id
+
+ response = self.session.post(url, headers=headers, json=data)
+ if response.status_code == 201: # Successful move returns 201 with empty body
+ success_message = f"Card {card_id} moved successfully to position {position}" + (f" in column {column_id}" if column_id else ".")
+ self.logger.info(success_message)
+ return success_message # Return success message as body is empty
+ else:
+ error_message = f"Error moving card {card_id}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ # _link_issue_to_project_card is effectively handled by _create_project_card if content_id and content_type are passed.
+ # The API used to have a separate link endpoint, but now it is part of card creation/update.
+ # For updating an existing card to link an issue, one would PATCH the card's content_id/content_type.
+ # Let's assume the function intends to update an existing card if it's a separate function.
+ # However, the provided API spec for `link_issue_to_project_card` uses PATCH on card_id, so let's implement that.
+ def _link_issue_to_project_card(self, card_id, content_id, content_type):
+ self.logger.info(f"Linking content_id {content_id} (type: {content_type}) to card_id {card_id}")
+ url = f"{self.base_url}/projects/cards/{card_id}" # Note: API docs suggest /projects/columns/cards/{card_id} or /projects/cards/{card_id}
+ # Using /projects/cards/{card_id} as it seems more general for card update.
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+ data = {"content_id": content_id, "content_type": content_type}
+
+ response = self.session.patch(url, headers=headers, json=data)
+ if response.status_code == 200:
+ updated_card = response.json()
+ success_message = f"{content_type} {content_id} linked to card {card_id} successfully."
+ self.logger.info(success_message)
+ return updated_card
+ else:
+ error_message = f"Error linking {content_type} {content_id} to card {card_id}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _list_project_boards(self):
+ self.logger.info(f"Listing project boards for repo: {self._repo}")
+ url = f"{self.base_url}/repos/{self._repo}/projects"
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+ response = self.session.get(url, headers=headers)
+ if response.status_code == 200:
+ projects_data = response.json()
+ self.logger.info(f"Successfully listed {len(projects_data)} project boards.")
+ return projects_data
+ else:
+ error_message = f"Error listing project boards: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _view_project_board_items(self, project_id):
+ self.logger.info(f"Viewing items for project ID: {project_id}")
+ columns_url = f"{self.base_url}/projects/{project_id}/columns"
+ headers = self.session.headers.copy()
+ headers["Accept"] = "application/vnd.github.inertia-preview+json"
+
+ columns_response = self.session.get(columns_url, headers=headers)
+ if columns_response.status_code != 200:
+ error_message = f"Error fetching columns for project {project_id}: {columns_response.status_code} - {columns_response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ columns_data = columns_response.json()
+ project_items = []
+ for column in columns_data:
+ column_info = {"id": column["id"], "name": column["name"], "cards": []}
+ cards_url = column["cards_url"]
+ cards_response = self.session.get(cards_url, headers=headers)
+ if cards_response.status_code == 200:
+ column_info["cards"] = cards_response.json()
+ else:
+ self.logger.error(f"Error fetching cards for column {column['id']}('{column['name']}'): {cards_response.status_code} - {cards_response.text}")
+ column_info["cards"] = "Error fetching cards"
+ project_items.append(column_info)
+
+ self.logger.info(f"Successfully retrieved items for project ID: {project_id}.")
+ return project_items
+
+ def _get_pull_request_details(self, pull_number):
+ self.logger.info(f"Getting details for PR #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ self.logger.info(f"Successfully retrieved details for PR #{pull_number}.")
+ return response.json()
+ else:
+ error_message = f"Error getting details for PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_pull_request_diff(self, pull_number):
+ self.logger.info(f"Getting diff for PR #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}"
+ diff_headers = self.session.headers.copy()
+ diff_headers["Accept"] = "application/vnd.github.diff"
+ response = self.session.get(url, headers=diff_headers)
+ if response.status_code == 200:
+ self.logger.info(f"Successfully retrieved diff for PR #{pull_number}.")
+ return response.text
+ else:
+ error_message = f"Error getting diff for PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _get_pull_request_files(self, pull_number):
+ self.logger.info(f"Getting files for PR #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/files"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ self.logger.info(f"Successfully retrieved files for PR #{pull_number}.")
+ return response.json()
+ else:
+ error_message = f"Error getting files for PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _create_pull_request_review_comment(self, pull_number, body, commit_id, path, position, side="RIGHT", start_line=None, start_side=None):
+ self.logger.info(f"Creating review comment on PR #{pull_number}, file '{path}', position {position}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments"
+ data = {"body": body, "commit_id": commit_id, "path": path, "position": position, "side": side}
+ if start_line is not None:
+ data["start_line"] = start_line
+ if start_side is not None:
+ data["start_side"] = start_side
+
+ response = self.session.post(url, json=data)
+ if response.status_code == 201:
+ comment_url = response.json().get("html_url", "N/A")
+ success_message = f"Review comment created successfully on PR #{pull_number}: {comment_url}"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error creating review comment on PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _list_pull_request_review_comments(self, pull_number):
+ self.logger.info(f"Listing review comments for PR #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments"
+ response = self.session.get(url)
+ if response.status_code == 200:
+ self.logger.info(f"Successfully retrieved review comments for PR #{pull_number}.")
+ return response.json()
+ else:
+ error_message = f"Error listing review comments for PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
+
+ def _submit_pull_request_review(self, pull_number, event, body=None):
+ self.logger.info(f"Submitting '{event}' review for PR #{pull_number}")
+ url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews"
+ data = {"event": event.upper()} # Ensure event is uppercase as per API
+ if body:
+ data["body"] = body
+
+ response = self.session.post(url, json=data)
+ if response.status_code == 200:
+ review_url = response.json().get("html_url", "N/A")
+ success_message = f"Review ({event}) submitted successfully for PR #{pull_number}: {review_url}"
+ self.logger.info(success_message)
+ return success_message
+ else:
+ error_message = f"Error submitting review for PR #{pull_number}: {response.status_code} - {response.text}"
+ self.logger.error(error_message)
+ return error_message
\ No newline at end of file
diff --git a/tools/standalone_llm_tool.py b/tools/standalone_llm_tool.py
index 41d8c0d..6df2e0b 100644
--- a/tools/standalone_llm_tool.py
+++ b/tools/standalone_llm_tool.py
@@ -3,6 +3,7 @@ import os
import json
import logging
from openai import OpenAI
+import re
import urllib.request
import urllib.error
@@ -82,10 +83,12 @@ class StandaloneLLMTool(BaseTool):
headers={'Content-Type': 'text/plain; charset=utf-8', 'User-Agent': 'DualAICopilot/0.1'},
method='POST'
)
- with urllib.request.urlopen(req, timeout=500) as response:
+ with urllib.request.urlopen(req, timeout=3600) as response:
if response.status == 200:
response_data = response.read().decode('utf-8')
logging.info(f"Received response from external copilot: {response_data[:100]}...")
+ # Remove content within tags
+ response_data = re.sub(r".*?", "", response_data, flags=re.DOTALL)
return response_data
else:
error_message = f"External copilot at {self.copilot_url} returned an error: {response.status} {response.reason}"