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.

Why Your AI Agent Workflow Needs Better Automation Tools in 2026

TL;DR: AI agent workflow automation tools in 2026 are no longer optional—they’re the difference between a proof-of-concept that… ...

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.

Build a High-Performance Async Web Scraper in Python: A Step-by-Step Tutorial

Build a High-Performance Async Web Scraper in Python: A Step-by-Step Tutorial Web scraping at scale is a classic… ...

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-sdk

Now, 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 issues

A 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.