How to Build an AI Code Reviewer That Remembers Team Standards

import os

ARCADE_USER_AGENT = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/126.0 Safari/537.36"
)

arcade_server = {
    "type": "http",
    "url": f"https://api.arcade.dev/mcp/{os.environ['ARCADE_GATEWAY_SLUG']}",
    "alwaysLoad": True,
    "headers": {
        "Authorization": f"Bearer {os.environ['ARCADE_API_KEY']}",
        "Arcade-User-ID": os.environ["ARCADE_USER_ID"],
        "User-Agent": ARCADE_USER_AGENT,
    },
}

# check_arcade_connection.py

import json
import os
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path

from dotenv import load_dotenv

def post_json(url: str, payload: dict, headers: dict[str, str]) -> tuple[int, dict, bytes]:
    request = urllib.request.Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers=headers,
        method="POST",
    )
    context = ssl.create_default_context()
    with urllib.request.urlopen(request, timeout=30, context=context) as response:
        body = response.read()
        return response.status, dict(response.headers.items()), body

def main() -> int:
    env_path = Path(__file__).with_name(".env")
    load_dotenv(env_path)

required = ("ARCADE_API_KEY", "ARCADE_USER_ID", "ARCADE_GATEWAY_SLUG")
    missing = [name for name in required if not os.environ.get(name)]
    if missing:
        print(f"Missing required environment variables: {', '.join(missing)}")
        return 1

gateway_url = f"https://api.arcade.dev/mcp/{os.environ['ARCADE_GATEWAY_SLUG']}"
    common_headers = {
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json",
        "Authorization": f"Bearer {os.environ['ARCADE_API_KEY']}",
        "Arcade-User-ID": os.environ["ARCADE_USER_ID"],
        "MCP-Protocol-Version": "2025-06-18",
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/126.0 Safari/537.36",
    }

initialize_payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2025-06-18",
            "capabilities": {},
            "clientInfo": {
                "name": "arcade-connection-check",
                "version": "1.0.0",
            },
        },
    }

try:
        print(f"Connecting to {gateway_url}")
        status, response_headers, body = post_json(
            gateway_url,
            initialize_payload,
            common_headers,
        )
        print(f"Initialize HTTP status: {status}")
        print(f"Response body: {body.decode('utf-8', errors='replace')}")

session_id = response_headers.get("Mcp-Session-Id")
        if not session_id:
            print("No Mcp-Session-Id was returned. The gateway did not start an MCP session.")
            return 1

print(f"MCP session established: {session_id}")

tools_payload = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {},
        }
        tools_headers = {
            **common_headers,
            "Mcp-Session-Id": session_id,
        }
        status, _, body = post_json(gateway_url, tools_payload, tools_headers)
        print(f"tools/list HTTP status: {status}")

parsed = json.loads(body.decode("utf-8"))
        tools = parsed.get("result", {}).get("tools", [])
        print(f"Gateway returned {len(tools)} tool(s).")
        for tool in tools:
            print(f"- {tool.get('name')}")

if not tools:
            print("Connected, but the gateway returned no tools.")
            return 1

print("Arcade MCP gateway connection looks healthy.")
        return 0

except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", errors="replace")
        print(f"HTTP error: {exc.code} {exc.reason}")
        print(error_body)
        if "browser_signature_banned" in error_body:
            print(
                "The request was blocked by Cloudflare before it reached Arcade. "
                "This usually means the HTTP client fingerprint was denied."
            )
        return 1
    except urllib.error.URLError as exc:
        print(f"Network error: {exc.reason}")
        return 1
    except json.JSONDecodeError as exc:
        print(f"Could not parse JSON response: {exc}")
        return 1
    except Exception as exc:
        print(f"Unexpected error: {type(exc).__name__}: {exc}")
        return 1

if __name__ == "__main__":
    sys.exit(main())

# authorize.py

import os
import time
from arcadepy import Arcade
from dotenv import load_dotenv

load_dotenv()

client = Arcade(api_key=os.environ["ARCADE_API_KEY"])
user_id = os.environ["ARCADE_USER_ID"]
tools = (
    "Github.GetPullRequest",
    "Github.CreateReviewComment",
    "Github.SubmitPullRequestReview",
)
AUTH_TIMEOUT_SECONDS = 300

for tool_name in tools:
    response = client.tools.authorize(tool_name=tool_name, user_id=user_id)
    if response.status == "completed":
        continue
    if response.status == "failed":
        raise RuntimeError(f"Authorization failed for {tool_name}.")
    if not response.id or not response.url:
        raise RuntimeError(
            f"Arcade did not return an authorization ID and URL for {tool_name}."
        )

print(f"Authorize {tool_name}: {response.url}")
    deadline = time.monotonic() + AUTH_TIMEOUT_SECONDS
    while response.status not in {"completed", "failed"}:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError(
                f"Authorization for {tool_name} did not finish within "
                f"{AUTH_TIMEOUT_SECONDS} seconds."
            )
        response = client.auth.status(
            id=response.id,
            wait=min(45, max(1, int(remaining))),
        )

if response.status == "failed":
        raise RuntimeError(f"Authorization failed for {tool_name}.")

print("GitHub authorization complete.")

# ingest.py

import json
import os
import time

from dotenv import load_dotenv
from hydra_db import HydraDB
from hydra_db.errors import ConflictError

load_dotenv()

DATABASE = "acme_code_review"
COLLECTION = "backend_team"

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])

def wait_for_database(timeout: int = 300) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        status = client.databases.status(database=DATABASE)
        if status.data.infra.ready_for_ingestion:
            print("Database ready.")
            return
        print("Waiting for database provisioning...")
        time.sleep(2)
    raise TimeoutError(f"Database {DATABASE!r} was not ready within {timeout} seconds.")

def wait_for_indexing(source_ids: list[str], timeout: int = 300) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        response = client.context.status(
            database=DATABASE,
            collection=COLLECTION,
            ids=source_ids,
        )
        statuses = response.data.statuses or []
        states = {item.id: item.indexing_status for item in statuses}

if any(state in {"failed", "errored"} for state in states.values()):
            errors = {
                item.id: item.error_message or item.message
                for item in statuses
                if item.indexing_status in {"failed", "errored"}
            }
            raise RuntimeError(f"Indexing failed: {errors}")

if len(states) == len(source_ids) and all(
            state == "completed" for state in states.values()
        ):
            print("Standards indexed and graph-ready.")
            return

print(f"Indexing status: {states}")
        time.sleep(3)

raise TimeoutError(f"Indexing did not finish within {timeout} seconds.")

# Create the database if needed.

try:
    client.databases.create(database=DATABASE)
    print(f"Created database {DATABASE!r}.")
except ConflictError:
    print(f"Database {DATABASE!r} already exists; reusing it.")

wait_for_database()

# Ingest the team standards

with open("standards.md", "rb") as f:
    result = client.context.ingest(
        database=DATABASE,
        collection=COLLECTION,
        type="knowledge",
        documents=("standards.md", f, "text/markdown"),
        document_metadata=json.dumps([{
            "id": "backend-team-standards-v1",
            "title": "Backend team coding standards",
        }]),
        upsert="true",
    )

source_ids = [item.id for item in result.data.results]
if not source_ids:
    raise RuntimeError("HydraDB did not return a source ID for the standards.")
print(f"Queued standards ingestion. Source IDs: {source_ids}")
wait_for_indexing(source_ids)

# Smoke-test retrieval before moving on to the agent.

smoke_test = client.query(
    database=DATABASE,
    collection=COLLECTION,
    query="What is the team's standard for nested conditionals and early returns?",
    type="knowledge",
    query_by="hybrid",
    mode="fast",
    graph_context=False,
    max_results=3,
)
if not (smoke_test.data.chunks or []):
    raise RuntimeError("Smoke test returned no standards. Check the ingestion status.")
print("Smoke test passed: the standards are queryable.")

# hydra_tools.py

import asyncio
import hashlib
import json
import os
import time
from typing import Any

from claude_agent_sdk import tool, create_sdk_mcp_server
from hydra_db import HydraDB
from hydra_db.helpers import build_string

DATABASE = "acme_code_review"
TEAM_COLLECTION = "backend_team"

client = HydraDB(token=os.environ["HYDRA_DB_API_KEY"])

def repo_collection(owner: str, repo: str) -> str:
    """Return a deterministic, collection-safe scope for one GitHub repository."""
    repo_key = f"{owner}/{repo}".lower().encode("utf-8")
    return f"repo_{hashlib.sha256(repo_key).hexdigest()[:16]}"

async def wait_for_indexing(
    source_ids: list[str],
    collection: str,
    timeout: int = 300,
) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        response = client.context.status(
            database=DATABASE,
            collection=collection,
            ids=source_ids,
        )
        statuses = response.data.statuses or []
        states = {item.id: item.indexing_status for item in statuses}

if any(state in {"failed", "errored"} for state in states.values()):
            errors = {
                item.id: item.error_message or item.message
                for item in statuses
                if item.indexing_status in {"failed", "errored"}
            }
            raise RuntimeError(f"Memory indexing failed: {errors}")

if len(states) == len(source_ids) and all(
            state == "completed" for state in states.values()
        ):
            return

await asyncio.sleep(2)

raise TimeoutError(f"Memory indexing did not finish within {timeout} seconds.")

@tool(
    "search_team_context",
    "Search team coding standards plus review history for one repository and "
    "author. Use type='knowledge' for standards only, 'memory' for past reviews "
    "only, or 'all' for both.",
    {"owner": str, "repo": str, "author": str, "query": str, "type": str},
)
async def search_team_context(args: dict[str, Any]) -> dict[str, Any]:
    repository = f"{args['owner']}/{args['repo']}"
    repository_scope = repo_collection(args["owner"], args["repo"])
    collection_response = client.databases.collections(database=DATABASE)
    available_collections = set(
        (collection_response.data.collections if collection_response.data else []) or []
    )
    query_collections = [TEAM_COLLECTION]
    if repository_scope in available_collections:
        query_collections.append(repository_scope)

query_text = (
        f"Repository: {repository}\n"
        f"Author: {args['author']}\n"
        f"{args['query']}"
    )
    result = client.query(
        query=query_text,
        database=DATABASE,
        collections=query_collections,
        type=args["type"],
        query_by="hybrid",
        mode="thinking",
        max_results=8,
        graph_context=True,
    )

context = build_string(result)
    return {"content": [{"type": "text", "text": context}]}

@tool(
    "store_review_memory",
    "Store a completed review in the repository's scoped memory. Include only "
    "verified review facts: the actual author, files and functions reviewed, "
    "issues found, and standards cited.",
    {
        "owner": str,
        "repo": str,
        "pr_number": int,
        "author": str,
        "summary": str,
    },
)
async def store_review_memory(args: dict[str, Any]) -> dict[str, Any]:
    repository = f"{args['owner']}/{args['repo']}"
    collection = repo_collection(args["owner"], args["repo"])
    review_id = f"pr_{args['pr_number']}_review"
    title = f"{repository} PR #{args['pr_number']} review"

result = client.context.ingest(
        database=DATABASE,
        collection=collection,
        type="memory",
        memories=json.dumps(
            [{
                "id": review_id,
                "title": title,
                "text": (
                    f"Repository: {repository}\n"
                    f"Pull request: #{args['pr_number']}\n"
                    f"Author: {args['author']}\n\n"
                    f"{args['summary']}"
                ),
                "is_markdown": True,
                "infer": False,
                "additional_metadata": {
                    "repository": repository,
                    "pr_number": args["pr_number"],
                    "author": args["author"],
                },
            }]
        ),
        upsert="true",
    )

source_ids = [item.id for item in result.data.results]
    if not source_ids:
        raise RuntimeError("HydraDB did not return a source ID for the review memory.")
    await wait_for_indexing(source_ids, collection)
    return {
        "content": [{
            "type": "text",
            "text": f"Stored and indexed review memory: {title}",
        }]
    }

# Package both tools into an in-process MCP server

hydra_server = create_sdk_mcp_server(
    name="hydra-context",
    version="1.0.0",
    tools=[search_team_context, store_review_memory],
)