← Back to DevBytes

Building a Code Review Agent with MCP (Model Context Protocol): Complete Guide

Building a Code Review Agent with MCP (Model Context Protocol): Complete Guide

Code review is one of the most valuable yet time-consuming activities in modern software development. With the rise of large language models (LLMs), automated code review has become increasingly viable — but wiring an LLM to your repository, CI pipeline, and review tooling has traditionally been a messy, bespoke effort. The Model Context Protocol (MCP) changes that by providing a standardized way for AI agents to access tools, resources, and context.

In this guide, you'll learn what MCP is, why it's a great fit for building a code review agent, and how to build one from scratch that can read diffs, analyze code, post comments, and integrate with your existing workflow.

What Is the Model Context Protocol (MCP)?

MCP is an open specification introduced by Anthropic that defines a standard protocol for connecting AI models to external data sources and tools. Think of it as the "USB-C of AI integrations" — a universal interface that lets any MCP-compatible client (such as Claude Desktop, an IDE extension, or a custom agent) talk to any MCP server that exposes capabilities.

An MCP server can expose three kinds of capabilities:

By standardizing these primitives, MCP lets you build an agent once and run it against many different clients and backends. For a code review agent, this means you can expose your Git provider, linters, and documentation as MCP tools and let any compatible model consume them.

Why MCP Matters for Code Review

Traditional automated code review tools fall into two camps: rigid rule-based linters that catch syntax issues but miss semantic problems, and ad-hoc LLM scripts that require custom glue code for every integration. MCP offers a middle path:

Architecture of a Code Review Agent

Our agent will consist of three layers:

We'll use Python with the official mcp SDK and the Anthropic SDK, but the same patterns apply in TypeScript or other languages.

Prerequisites and Setup

Before you begin, make sure you have:

Create a project directory and install dependencies:

mkdir code-review-agent && cd code-review-agent
python -m venv .venv
source .venv/bin/activate
pip install mcp anthropic httpx python-dotenv

Create a .env file to hold your secrets:

ANTHROPIC_API_KEY=sk-ant-...
GITHUB_TOKEN=ghp_...
GITHUB_REPOSITORY=your-org/your-repo

Building the MCP Server

The MCP server is the heart of the agent. It exposes the tools the model will call during a review. We'll implement four tools: get_pull_request, get_pr_diff, run_linter, and post_review_comment.

Create server.py:

import os
import json
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
REPO = os.environ["GITHUB_REPOSITORY"]
GITHUB_API = "https://api.github.com"

server = Server("code-review-server")

def gh_headers():
    return {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_pull_request",
            description="Fetch metadata for a pull request by number.",
            inputSchema={
                "type": "object",
                "properties": {
                    "pr_number": {"type": "integer", "description": "PR number"}
                },
                "required": ["pr_number"],
            },
        ),
        Tool(
            name="get_pr_diff",
            description="Fetch the unified diff of a pull request.",
            inputSchema={
                "type": "object",
                "properties": {
                    "pr_number": {"type": "integer", "description": "PR number"}
                },
                "required": ["pr_number"],
            },
        ),
        Tool(
            name="run_linter",
            description="Run a linter (e.g. flake8) on a file path in the repo.",
            inputSchema={
                "type": "object",
                "properties": {
                    "file_path": {"type": "string", "description": "Path to file"}
                },
                "required": ["file_path"],
            },
        ),
        Tool(
            name="post_review_comment",
            description="Post a review comment on a pull request.",
            inputSchema={
                "type": "object",
                "properties": {
                    "pr_number": {"type": "integer"},
                    "body": {"type": "string", "description": "Markdown comment body"}
                },
                "required": ["pr_number", "body"],
            },
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_pull_request":
        pr_number = arguments["pr_number"]
        async with httpx.AsyncClient() as client:
            r = await client.get(
                f"{GITHUB_API}/repos/{REPO}/pulls/{pr_number}",
                headers=gh_headers(),
            )
            r.raise_for_status()
            pr = r.json()
        summary = {
            "title": pr["title"],
            "author": pr["user"]["login"],
            "state": pr["state"],
            "body": pr.get("body", ""),
            "changed_files": pr["changed_files"],
            "additions": pr["additions"],
            "deletions": pr["deletions"],
        }
        return [TextContent(type="text", text=json.dumps(summary, indent=2))]

    elif name == "get_pr_diff":
        pr_number = arguments["pr_number"]
        async with httpx.AsyncClient() as client:
            r = await client.get(
                f"{GITHUB_API}/repos/{REPO}/pulls/{pr_number}",
                headers={**gh_headers(), "Accept": "application/vnd.github.diff"},
            )
            r.raise_for_status()
        return [TextContent(type="text", text=r.text)]

    elif name == "run_linter":
        file_path = arguments["file_path"]
        import subprocess
        result = subprocess.run(
            ["flake8", "--max-line-length=100", file_path],
            capture_output=True,
            text=True,
        )
        output = result.stdout + result.stderr
        return [TextContent(type="text", text=output or "No lint issues found.")]

    elif name == "post_review_comment":
        pr_number = arguments["pr_number"]
        body = arguments["body"]
        async with httpx.AsyncClient() as client:
            r = await client.post(
                f"{GITHUB_API}/repos/{REPO}/issues/{pr_number}/comments",
                headers=gh_headers(),
                json={"body": body},
            )
            r.raise_for_status()
        return [TextContent(type="text", text="Comment posted successfully.")]

    else:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

This server uses the stdio transport, which is the simplest way to run an MCP server as a subprocess of your agent. Each tool has a JSON Schema describing its inputs, which the model uses to decide when and how to call it.

Building the Agent Loop

The agent loop is responsible for driving the conversation with the LLM. It sends the available tools, receives tool-call requests from the model, executes them against the MCP server, and feeds the results back until the model produces a final review.

Create agent.py:

import os
import json
import asyncio
from contextlib import AsyncExitStack
from dotenv import load_dotenv
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

load_dotenv()

SYSTEM_PROMPT = """You are a senior code reviewer. When given a pull request number:
1. Fetch the PR metadata and diff using the available tools.
2. Identify files that warrant review and run the linter on them.
3. Analyze the diff for bugs, security issues, style violations, and improvements.
4. Post a single, well-structured review comment summarizing your findings.
Be concise, specific, and reference file paths and line numbers when possible.
If the change looks good, say so clearly and post a brief approval comment."""

class CodeReviewAgent:
    def __init__(self):
        self.anthropic = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
        self.session = None
        self.exit_stack = AsyncExitStack()
        self.model = "claude-3-5-sonnet-20241022"

    async def connect(self):
        server_params = StdioServerParameters(
            command="python",
            args=["server.py"],
            env={
                "GITHUB_TOKEN": os.environ["GITHUB_TOKEN"],
                "GITHUB_REPOSITORY": os.environ["GITHUB_REPOSITORY"],
                "PATH": os.environ["PATH"],
            },
        )
        stdio_transport = await self.exit_stack.enter_async_context(
            stdio_client(server_params)
        )
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.stdio, self.write)
        )
        await self.session.initialize()

    async def review(self, pr_number: int) -> str:
        tools_result = await self.session.list_tools()
        tools = [
            {
                "name": t.name,
                "description": t.description,
                "input_schema": t.inputSchema,
            }
            for t in tools_result.tools
        ]

        messages = [
            {
                "role": "user",
                "content": f"Please review pull request #{pr_number}.",
            }
        ]

        while True:
            response = self.anthropic.messages.create(
                model=self.model,
                max_tokens=4096,
                system=SYSTEM_PROMPT,
                tools=tools,
                messages=messages,
            )

            if response.stop_reason == "end_turn":
                final_text = "".join(
                    block.text for block in response.content if block.type == "text"
                )
                return final_text

            assistant_content = []
            for block in response.content:
                if block.type == "text":
                    assistant_content.append({"type": "text", "text": block.text})
                elif block.type == "tool_use":
                    assistant_content.append(
                        {
                            "type": "tool_use",
                            "id": block.id,
                            "name": block.name,
                            "input": block.input,
                        }
                    )
            messages.append({"role": "assistant", "content": assistant_content})

            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  -> Calling tool: {block.name}({block.input})")
                    result = await self.session.call_tool(block.name, block.input)
                    tool_text = "\n".join(c.text for c in result.content)
                    tool_results.append(
                        {
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": tool_text,
                        }
                    )
            messages.append({"role": "user", "content": tool_results})

    async def cleanup(self):
        await self.exit_stack.aclose()

async def main():
    import sys
    if len(sys.argv) < 2:
        print("Usage: python agent.py <pr_number>")
        return
    pr_number = int(sys.argv[1])
    agent = CodeReviewAgent()
    try:
        await agent.connect()
        print(f"Reviewing PR #{pr_number}...")
        result = await agent.review(pr_number)
        print("\n=== Review Summary ===\n")
        print(result)
    finally:
        await agent.cleanup()

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

The loop continues until the model signals end_turn, meaning it has finished its analysis and (in our case) posted a review comment via the post_review_comment tool. Each tool call is executed against the live MCP session, and the result is appended to the conversation so the model can reason about it.

Running the Agent

With both files in place, run the agent against a real pull request:

export $(cat .env | xargs)
python agent.py 42

You should see output like:

Reviewing PR #42...
  -> Calling tool: get_pull_request({'pr_number': 42})
  -> Calling tool: get_pr_diff({'pr_number': 42})
  -> Calling tool: run_linter({'file_path': 'src/auth.py'})
  -> Calling tool: post_review_comment({'pr_number': 42, 'body': '## Code Review ...'})

=== Review Summary ===

I reviewed PR #42 "Add OAuth login flow" and posted a comment covering ...

The agent autonomously fetched the PR, inspected the diff, ran the linter on a changed file, and posted a structured review comment back to GitHub.

Adding Resources for Context

Tools are great for actions, but a good reviewer also needs context. MCP resources let you expose read-only data like coding standards, architecture docs, or historical review patterns. Let's add a resource that serves the repository's style guide.

Add this to server.py:

from mcp.types import Resource

STYLE_GUIDE = """
# Repository Style Guide
- Use type hints on all public functions.
- Keep functions under 50 lines; extract helpers when longer.
- Prefer dataclasses over dicts for structured data.
- Never commit secrets; use environment variables.
- All new endpoints must have integration tests.
"""

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="style-guide://repo",
            name="Repository Style Guide",
            description="Coding standards for this repository.",
            mimeType="text/markdown",
        )
    ]

@server.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "style-guide://repo":
        return STYLE_GUIDE
    raise ValueError(f"Unknown resource: {uri}")

Then update the agent to read the style guide before reviewing:

style_guide = await self.session.read_resource("style-guide://repo")
messages[0]["content"] += f"\n\nStyle guide:\n{style_guide}"

Now the agent's reviews will be grounded in your team's actual standards rather than generic best practices.

Best Practices

Extending the Agent

Once the core works, you can extend it in several directions:

Conclusion

Building a code review agent with MCP gives you a clean, standardized, and extensible foundation for automating one of software engineering's most labor-intensive tasks. By separating concerns — tools on the server, orchestration in the agent loop, and reasoning in the model — you get a system that's easy to debug, simple to extend, and portable across model providers. Start small with the four tools we built here, iterate on the prompts and resources, and gradually add capabilities as your team's needs grow. With thoughtful tool design and clear guardrails, an MCP-based review agent can meaningfully reduce review latency while freeing human reviewers to focus on the architectural and domain-specific feedback that matters most.

— Ad —

Google AdSense will appear here after approval

← Back to all articles