GitHub PR Review Agent

{
  "organization_id": "<YOUR_ORG_ID>",
  "api_key": "<YOUR_XPANDER_API_KEY>",
  "agent_id": "<YOUR_AGENT_ID>"
}

from agno.tools import tool
from github import Github
from core.pr_review import summarize_diff, score_pr, generate_action_items, final_decision
from core.slack import send_slack_message
import os

@tool(
    name="PRReviewTool",
    description="Analyze a GitHub PR, generate summary, score, and auto-merge if approved.",
    show_result=True,
    stop_after_tool_call=True
)
def review_pr(pr_url: str) -> dict:
    github_token = os.getenv("GITHUB_TOKEN")
    
    # Connect to GitHub and fetch PR data
    g = Github(github_token)
    repo_name, pr_number = extract_repo_and_number(pr_url)
    repo = g.get_repo(repo_name)
    pr = repo.get_pull(pr_number)
    
    # Analyze the PR
    summary = summarize_diff(pr.get_files())
    score = score_pr(pr)
    actions = generate_action_items(pr)
    decision = final_decision(score)
    
    # Auto-merge if approved
    merge_status = "Skipped"
    if decision == "✅ Approve":
        try:
            pr.merge(commit_message="Auto-merged by PR Review Agent ✅")
            merge_status = "Success"
        except Exception as e:
            merge_status = f"Failed: {str(e)}"
    
    # Send Slack notification
    slack_msg = format_slack_message(repo_name, pr_number, pr.title, score, summary, actions, decision, merge_status)
    send_slack_message(slack_msg)
    
    return {
        "summary": summary,
        "score": score,
        "decision": decision,
        "merge_status": merge_status
    }

import sqlite3

def summarize_diff(pr_files) -> str:
    if not pr_files:
        return "No files changed in this PR."
    
    summary_lines = []
    for f in pr_files:
        summary_lines.append(
            f"📄 **{f.filename}**\n"
            f"- ➕ {f.additions} additions\n"
            f"- ➖ {f.deletions} deletions\n"
        )
    return "\n".join(summary_lines)

def score_pr(pr):
    score = 10
    if pr.additions > 500:
        score -= 2
    if pr.deletions > pr.additions:
        score -= 1
    if not pr.body:
        score -= 1
    return max(score, 0)

def generate_action_items(pr):
    items = []
    if "fix" in pr.title.lower() and "test" not in pr.title.lower():
        items.append("Add test cases for the fix.")
    if pr.additions > 1000:
        items.append("Break the PR into smaller parts.")
    return items

def final_decision(score):
    return "✅ Approve" if score >= 7 else "❌ Push Back"

import os
import requests
import json

def send_slack_message(text: str):
    slack_token = os.getenv("SLACK_BOT_TOKEN")
    slack_channel = os.getenv("SLACK_CHANNEL_ID")
    
    headers = {
        "Authorization": f"Bearer {slack_token}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "channel": slack_channel,
        "text": text,
        "unfurl_links": False
    }
    
    response = requests.post(
        "https://slack.com/api/chat.postMessage",
        headers=headers,
        data=json.dumps(payload)
    )
    
    return response.json()

from agno.agent import Agent
from agno.models.nebius import Nebius
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory
from agno.storage.sqlite import SqliteStorage
from agno.tools.thinking import ThinkingTools
from tools.pr_tools import review_pr
import os

class PRReviewOrchestrator:
    def __init__(self, agent_backend):
        self.agent_backend = agent_backend
        self.agent = None
        self.tools = [ThinkingTools(add_instructions=True), review_pr]
        
        # Set up memory and storage
        self.memory = Memory(
            model=Nebius(id="Qwen/Qwen3-235B-A22B", api_key=os.getenv("NEBIUS_API_KEY")),
            db=SqliteMemoryDb(table_name="user_memories", db_file="agent_state.db")
        )
        self.storage = SqliteStorage(table_name="agent_sessions", db_file="agent_state.db")
    
    async def run(self, message: str, user_id: str, session_id: str, cli: bool = False):
        if not self.agent:
            self.agent = Agent(
                model=Nebius(id="Qwen/Qwen3-235B-A22B", api_key=os.getenv("NEBIUS_API_KEY")),
                tools=self.tools,
                memory=self.memory,
                storage=self.storage,
                session_id=session_id,
                user_id=user_id
            )
        
        return await self.agent.arun(message, stream=False)

import json
import asyncio
from xpander_utils.events import XpanderEventListener, AgentExecutionResult
from xpander_utils.sdk.adapters import AgnoAdapter
from orchestrator.agno_agent import PRReviewOrchestrator
from pathlib import Path

# Load configuration
CFG_PATH = Path("xpander_config.json")
xpander_cfg = json.loads(CFG_PATH.read_text())

# Initialize backend and orchestrator
xpander_backend = asyncio.run(
    asyncio.to_thread(AgnoAdapter, agent_id=xpander_cfg["agent_id"], api_key=xpander_cfg["api_key"])
)
agno_agent_orchestrator = PRReviewOrchestrator(xpander_backend)

async def on_execution_request(execution_task) -> AgentExecutionResult:
    try:
        await asyncio.to_thread(xpander_backend.agent.init_task, execution=execution_task.model_dump())
        
        agent_response = await agno_agent_orchestrator.run(
            execution_task.input.text,
            execution_task.input.user.id,
            execution_task.memory_thread_id
        )
        
        return AgentExecutionResult(result=agent_response.content, is_success=True)
    except Exception as e:
        return AgentExecutionResult(result=str(e), is_success=False)

# Register the handler
listener = XpanderEventListener(**xpander_cfg)
listener.register(on_execution_request=on_execution_request)

import asyncio
import json
from pathlib import Path
from orchestrator.agno_agent import PRReviewOrchestrator
from xpander_utils.sdk.adapters import AgnoAdapter

async def main():
    cfg_path = Path("xpander_config.json")
    xpander_cfg = json.loads(cfg_path.read_text())
    
    xpander_backend = AgnoAdapter(
        agent_id=xpander_cfg["agent_id"],
        api_key=xpander_cfg["api_key"]
    )
    
    orchestrator = PRReviewOrchestrator(xpander_backend)
    
    while True:
        message = input("\nYou: ")
        if message.lower() in ["exit", "quit"]:
            break
        
        response = await orchestrator.run(
            message=message,
            user_id="user123",
            session_id="session456",
            cli=True
        )
        
        print(f"Agent: {response.content}")

if __name__ == "__main__":
    asyncio.run(main())