Build a Custom AI Code Review Agent: A Step-by-Step Tutorial with ECOA AI Platform ACP
Let’s be honest. Code reviews are the bottleneck nobody wants to talk about.
You know the drill. A PR sits for 12 hours waiting for a senior dev to find time. When they finally look at it, they catch the same three types of bugs — null checks, missing error handling, and formatting inconsistencies. Meanwhile, the subtle architectural issues slip through.
I’ve been there. We all have.
But what if you could build your own AI code review agent that catches those issues *before* a human ever looks at the code? Not a generic linter. Not Copilot’s inline suggestions. A real agent that understands your project’s conventions, your team’s coding standards, and your specific business logic.
Building a Remote Engineering Team in Vietnam: A Step-by-Step Guide
Everything you need to know about building an effective remote engineering team in Vietnam. From hiring practices to… ...
That’s exactly what I’m going to show you today. We’ll build a custom AI code review agent using the ECOA AI Platform ACP (Agent Control Plane). This isn’t theory — it’s the same approach our team in Ho Chi Minh City used to cut review cycles by 60% for a fintech client last quarter.
Here’s what we’ll cover:
- Why a custom agent beats off-the-shelf tools
- The stack: Python + ECOA ACP + Claude API
- Step-by-step code walkthrough
- Real deployment configs that actually work
- Lessons learned from production
Let’s get into it.
Why Build a Custom Agent Instead of Using Copilot or PR-Agent?
Good question. Actually, it’s the first question my clients ask.
Off-the-shelf tools are great for generic issues. They’ll catch a missing semicolon or a potential SQL injection. But they don’t know your codebase. They don’t know that your team uses `snake_case` for variables but `PascalCase` for class names. They don’t know that you’ve decided to ban `any` types in TypeScript unless explicitly approved.
A custom agent does.
Here’s the real kicker: you control the review rules. Want the agent to flag any function over 50 lines? Done. Want it to check that every API endpoint has a rate limit? Easy. Want it to verify that all database queries use parameterized inputs? That’s a 5-line rule.
Plus, you can integrate it directly into your CI/CD pipeline. No third-party SaaS. No per-seat pricing. Just your code, your rules, your infrastructure.
Our team in Can Tho built a version that checks for Vietnamese character encoding issues in localization files. Try finding that in a generic tool.
The Stack: What You’ll Need
Before we dive into code, here’s the stack we’re using:
- ECOA AI Platform ACP: The orchestration layer that manages agent lifecycle, task routing, and error recovery
- Python 3.11+: The agent runtime
- Claude API (Sonnet 4): The LLM backend for code analysis
- GitHub Webhooks: To trigger reviews on new PRs
- Redis: For caching review results and avoiding duplicate runs
That’s it. No Kubernetes cluster. No complex microservices architecture. Just a single agent service that scales horizontally when needed.
Step 1: Setting Up the ECOA ACP Agent
First, install the ECOA ACP Python SDK. It’s a single pip command:
bash
pip install ecoa-acp-sdkNow, let’s create the agent configuration. This tells the ACP what kind of agent we’re building, what tools it has access to, and how it should behave.
python
# agent_config.py
from ecoa_acp import AgentConfig, ToolDefinition
config = AgentConfig(
agent_id="code-reviewer-v1",
description="AI code review agent that analyzes pull requests for bugs, style issues, and security vulnerabilities",
llm_config={
"provider": "anthropic",
"model": "claude-sonnet-4-20250514",
"max_tokens": 4096,
"temperature": 0.1 # Keep it deterministic for reviews
},
tools=[
ToolDefinition(
name="fetch_pr_diff",
description="Fetches the diff for a given PR from GitHub",
parameters={
"repo": {"type": "string", "description": "Repository name (e.g., owner/repo)"},
"pr_number": {"type": "integer", "description": "PR number"}
}
),
ToolDefinition(
name="fetch_file_content",
description="Fetches the full content of a specific file at a given commit",
parameters={
"repo": {"type": "string"},
"file_path": {"type": "string"},
"commit_sha": {"type": "string"}
}
),
ToolDefinition(
name="post_review_comment",
description="Posts a review comment on a specific line of a PR",
parameters={
"repo": {"type": "string"},
"pr_number": {"type": "integer"},
"body": {"type": "string", "description": "Comment text"},
"commit_id": {"type": "string"},
"path": {"type": "string", "description": "File path"},
"line": {"type": "integer", "description": "Line number"}
}
)
],
review_rules=[
"Flag any function exceeding 50 lines",
"Check that all database queries use parameterized inputs",
"Verify every public API endpoint has a rate limit annotation",
"Ensure no hardcoded secrets (API keys, passwords) in the diff",
"Flag any TypeScript 'any' type usage unless explicitly commented",
"Check for missing error handling in async functions"
]
)Notice the `temperature: 0.1`. For code review, you want consistency, not creativity. You don’t want the agent inventing issues that don’t exist.
Step 2: Implementing the Agent Logic
Now let’s write the actual agent. This is where the magic happens.
python
# code_review_agent.py
import os
import json
import hashlib
import redis
from typing import List, Dict
from github import Github
from ecoa_acp import AgentRuntime
from anthropic import Anthropic
# Initialize clients
redis_client = redis.Redis.from_url(os.environ["REDIS_URL"])
github_client = Github(os.environ["GITHUB_TOKEN"])
anthropic_client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
class CodeReviewAgent:
def __init__(self, config: dict):
self.runtime = AgentRuntime(config)
self.review_rules = config["review_rules"]
def generate_cache_key(self, repo: str, pr_number: int, commit_sha: str) -> str:
"""Generate a deterministic cache key to avoid re-reviewing unchanged code."""
raw = f"{repo}:{pr_number}:{commit_sha}"
return f"review:{hashlib.sha256(raw.encode()).hexdigest()}"
def fetch_pr_diff(self, repo: str, pr_number: int) -> str:
"""Fetch the PR diff from GitHub."""
repo_obj = github_client.get_repo(repo)
pr = repo_obj.get_pull(pr_number)
return pr.get_files() # Returns list of file objects with patch data
def analyze_diff(self, diff_data: List[Dict]) -> List[Dict]:
"""Send the diff to Claude for analysis based on our rules."""
prompt = self._build_review_prompt(diff_data)
response = anthropic_client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
temperature=0.1,
system="You are a senior code reviewer. Analyze the following diff and identify issues based on the provided rules. Return results as JSON.",
messages=[{"role": "user", "content": prompt}]
)
return json.loads(response.content[0].text)
def _build_review_prompt(self, diff_data: List[Dict]) -> str:
"""Build a structured prompt with the diff and rules."""
rules_text = "\n".join([f"- {rule}" for rule in self.review_rules])
diff_text = ""
for file in diff_data[:20]: # Limit to 20 files per review
diff_text += f"\n--- {file['filename']} ---\n"
diff_text += file.get('patch', '')[:5000] # Limit per file to avoid token overflow
return f"""Review the following pull request diff against these rules:
{rules_text}
Diff:
{diff_text}
Return a JSON array of issues found. Each issue should have:
- "file": the file path
- "line": the line number (integer)
- "severity": "error", "warning", or "info"
- "rule": which rule was violated
- "message": a clear explanation of the issue
- "suggestion": how to fix it
If no issues are found, return an empty array."""
def post_review(self, repo: str, pr_number: int, issues: List[Dict]):
"""Post review comments on the PR."""
repo_obj = github_client.get_repo(repo)
pr = repo_obj.get_pull(pr_number)
for issue in issues[:10]: # Limit to 10 comments to avoid noise
try:
pr.create_review_comment(
body=f"**{issue['severity'].upper()}**: {issue['message']}\n\n> Suggestion: {issue['suggestion']}",
commit_id=pr.get_commits()[0].sha,
path=issue['file'],
line=issue['line']
)
except Exception as e:
print(f"Failed to post comment on {issue['file']}:{issue['line']}: {e}")
def review_pr(self, repo: str, pr_number: int):
"""Main review entry point."""
pr = github_client.get_repo(repo).get_pull(pr_number)
latest_commit = pr.get_commits()[0].sha
# Check cache
cache_key = self.generate_cache_key(repo, pr_number, latest_commit)
cached = redis_client.get(cache_key)
if cached:
print(f"Using cached review for PR #{pr_number}")
return json.loads(cached)
# Fetch and analyze
diff_data = self.fetch_pr_diff(repo, pr_number)
issues = self.analyze_diff(diff_data)
# Cache for 24 hours
redis_client.setex(cache_key, 86400, json.dumps(issues))
# Post results
if issues:
self.post_review(repo, pr_number, issues)
return issuesA few things to note here:
- Caching is critical. Without it, you’ll re-review the same code every time someone pushes a new commit. We cache by commit SHA, so only new commits trigger a fresh review.
- We limit the diff size. Claude has a 200K token context window, but you don’t need to feed it every file. 20 files with 5K chars each is plenty for a meaningful review.
- Temperature stays at 0.1. I can’t stress this enough. You want deterministic, repeatable results.
Step 3: Deploying with GitHub Webhooks
The agent needs to be triggered automatically when a PR is opened or updated. Here’s the webhook handler:
python
# webhook_handler.py
from flask import Flask, request, jsonify
from code_review_agent import CodeReviewAgent
import os
app = Flask(__name__)
agent = CodeReviewAgent(config={
"review_rules": [
"Flag any function exceeding 50 lines",
"Check that all database queries use parameterized inputs",
# ... rest of rules
]
})
@app.route("/webhook/github", methods=["POST"])
def handle_github_webhook():
event = request.headers.get("X-GitHub-Event")
if event == "pull_request":
payload = request.json
action = payload.get("action")
# Only review on opened or synchronized (new commits)
if action in ["opened", "synchronize"]:
repo = payload["repository"]["full_name"]
pr_number = payload["pull_request"]["number"]
# Fire and forget - don't block the webhook response
from threading import Thread
Thread(target=agent.review_pr, args=(repo, pr_number)).start()
return jsonify({"status": "review_queued"}), 202
return jsonify({"status": "ignored"}), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)Deploy this behind a load balancer,
Related: software outsourcing services — Learn more about how ECOA AI can help your team.
Related: affordable software outsourcing — Learn more about how ECOA AI can help your team.
Related reading: Why Smart CTOs Hire Vietnamese Developers: A Data-Driven Guide for 2025
Related reading: Why Vietnam Outsourcing Is Winning: A No-Nonsense Guide for CTOs
