1 import os
2 from typing import Dict, List, Optional
3 import dspy
4 from dspy import Prediction
5 from agenspy import RealMCPClient
6
7 class PRReviewAgent(dspy.Module):
8 """Agent for automated PR reviews."""
9 
10 def __init__(self, mcp_client, github_token: Optional[str] = None):
11 super().__init__()
12 self.mcp_client = mcp_client
13 self.github_token = github_token
14
15 # Define predictors with modern signature syntax using ->
16 self.analyze_changes = dspy.Predict(
17 "pr_content -> security_issues: list[str], edge_cases: list[str]"
18 )
19 
20 self.review_docs = dspy.Predict(
21 "changes -> doc_updates: list[str], doc_suggestions: list[str]"
22 )
23 
24 self.analyze_impact = dspy.Predict(
25 "changes, codebase -> impact_analysis: str, risk_score: float"
26 )
27
28 def forward(self, pr_url: str) -> Prediction:
29 """Review a pull request and provide comprehensive analysis."""
30 print(f" Reviewing PR: {pr_url}")
31 
32 # Define default values
33 default_security = ["No security issues identified in the current codebase"]
34 default_edge_cases = ["No potential edge cases detected"]
35 default_doc_updates = ["No documentation updates required"]
36 default_doc_suggestions = ["Documentation appears to be complete"]
37 default_impact = "No significant impact detected"
38 default_risk = 0.0
39 
40 # Fetch PR data
41 pr_data = self.mcp_client(
42 context_request=f"Get PR details for {pr_url}",
43 tool_name="get_pull_request",
44 tool_args={"url": pr_url}
45 )
46
47 # Access the data properly from pr_data
48 context_data = pr_data.get('context_data', {})
49 diff_data = pr_data.get('diff', '') 
50
51 try:
52 # Analyze changes using modern predictor calls
53 analysis = self.analyze_changes(pr_content=context_data)
54 doc_review = self.review_docs(changes=diff_data)
55 impact = self.analyze_impact(
56 changes=diff_data,
57 codebase=context_data
58 )
59
60 # Return a Prediction object with defaults
61 return dspy.Prediction(
62 security_issues=getattr(analysis, 'security_issues', []) or default_security,
63 edge_cases=getattr(analysis, 'edge_cases', []) or default_edge_cases,
64 doc_updates=getattr(doc_review, 'doc_updates', []) or default_doc_updates,
65 doc_suggestions=getattr(doc_review, 'doc_suggestions', []) or default_doc_suggestions,
66 impact_analysis=getattr(impact, 'impact_analysis', '') or default_impact,
67 risk_score=getattr(impact, 'risk_score', 0.0) or default_risk
68 )
69 except Exception as e:
70 print(f"Analysis error: {str(e)}")
71 # Return prediction with default values on error
72 return dspy.Prediction(
73 security_issues=default_security,
74 edge_cases=default_edge_cases,
75 doc_updates=default_doc_updates,
76 doc_suggestions=default_doc_suggestions,
77 impact_analysis=default_impact,
78 risk_score=default_risk
79 )
80
81 def main():
82 """Run PR Review Agent demo."""
83 github_token = os.getenv('GITHUB_TOKEN')
84 openai_key = os.getenv('OPENAI_API_KEY')
85 
86 if not all([github_token, openai_key]):
87 raise EnvironmentError("Please set GITHUB_TOKEN and OPENAI_API_KEY")
88
89 # Configure DSPy with modern configuration
90 lm = dspy.LM('openai/gpt-4o-mini', api_key=openai_key)
91 dspy.configure(lm=lm)
92
93 # Setup MCP client
94 mcp_client = RealMCPClient(
95 ["npx", "-y", "@modelcontextprotocol/server-github"]
96 )
97
98 try:
99 # Create and run agent
100 agent = PRReviewAgent(mcp_client, github_token)
101 result = agent.forward("https://github.com/SuperagenticAI/SuperXLab/pull/1")
102
103 print("\n PR Review Results:")
104 for key, value in result.items():
105 print(f"\n{key.replace('_', ' ').title()}:")
106 if isinstance(value, list):
107 for item in value:
108 print(f" - {item}")
109 else:
110 print(f" {value}")
111 
112 except Exception as e:
113 print(f"Error during PR review: {str(e)}")
114
115if __name__ == "__main__":
116 main()