Practical notes: Building a Two-Agent Code Review Loop with A2A, LangGraph, and

The following notes reconstruct a practical path around “Building a Two-Agent Code Review Loop with A2A, LangGraph, and LM Studio”. Emphasis stays on contracts, checks, and drop-in code placeholders rather than motivational framing. When working through Overview, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

What A2A actually is (one paragraph)

What A2A actually is (one paragraph) works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.

The pieces

The pieces works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

a2a-coder-reviewer/
├── shared/
│   ├── config.py           # tunables: system prompts, model, temperature, MAX_ROUNDS
│   ├── llm.py              # LM Studio ChatOpenAI factory
│   ├── coder_graph.py      # LangGraph graph: spec (+ feedback) -> code
│   └── reviewer_graph.py   # LangGraph graph: spec + code -> verdict + feedback
├── coder_agent/
│   ├── agent_executor.py   # A2A executor wrapping the coder graph
│   └── __main__.py         # A2A server, port 9001
├── reviewer_agent/
│   ├── agent_executor.py   # A2A executor wrapping the reviewer graph
│   └── __main__.py         # A2A server, port 9002
├── orchestrator.py         # A2A client driving the back-and-forth
└── requirements.txt

The Config file

The Config file works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts. The Config file works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

from dataclasses import dataclass

@dataclass(frozen=True)
class AgentConfig:
    model: str
    temperature: float
    system_prompt: str


CODER_CONFIG = AgentConfig(
    model="local-model",
    temperature=0.2,
    system_prompt="""You are a careful Python engineer.
Write a single Python function that satisfies the spec.
If reviewer feedback is provided, revise your previous code to address it.
Return ONLY the code in a single ```python fenced block, no extra commentary.
""",
)

REVIEWER_CONFIG = AgentConfig(
    model="local-model",
    temperature=0.0,
    system_prompt="""You are a strict senior Python code reviewer performing a
rigorous code review. Do not approve code that has any of these problems:
- Missing a docstring explaining parameters, return value, and behavior.
- Missing type hints on parameters and the return type.
- Failing to handle edge cases implied by the spec (empty input, None,
  zero, negative numbers, single-element collections, etc.) unless the
  spec explicitly says not to.
- Silently swallowing errors or returning a sentinel value (None, "", -1,
  an error string) instead of raising, unless the spec asks for that.
- Using a data structure that visibly discards information the spec
  requires (e.g. a `set` when order or duplicate counts must be kept).

If you see ANY of the above, or anything else that would fail a real code
review, you MUST respond with VERDICT: CHANGES_REQUESTED and list the
concrete problems as bullet points, even if the code "basically works."
Only respond with VERDICT: APPROVED if the code has none of these
problems. Never approve code and then list problems as "suggestions" or
"nice to haves" — if it's worth mentioning, it's worth requesting changes
for.

Start your reply with exactly one line:
VERDICT: APPROVED
or
VERDICT: CHANGES_REQUESTED
Then, on the following lines, give concise, actionable feedback
(bullet points). If APPROVED, feedback can just be brief praise/notes.
""",
)

# Max coder<->reviewer rounds before orchestrator.py gives up.
MAX_ROUNDS = 3

The LLM factory

For The LLM factory, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Prefer structured outputs with schema validation over free-form prose when the next step is code or a tool call.

import os
from langchain_openai import ChatOpenAI

LM_STUDIO_BASE_URL = os.environ.get(
    "LM_STUDIO_BASE_URL", "http://localhost:7777/v1"
)

def get_llm(model: str = "local-model", temperature: float = 0.2) -> ChatOpenAI:
    return ChatOpenAI(
        base_url=LM_STUDIO_BASE_URL,
        api_key="lm-studio",       # placeholder, LM Studio ignores it
        model=model,
        temperature=temperature,
    )

The Coder graph

For The Coder graph, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
from shared.config import CODER_CONFIG
from shared.llm import get_llm

class CoderState(TypedDict):
    spec: str
    previous_code: Optional[str]
    feedback: Optional[str]
    code: str


def _generate(state: CoderState) -> CoderState:
    llm = get_llm(model=CODER_CONFIG.model, temperature=CODER_CONFIG.temperature)
    user_msg = f"Spec:\n{state['spec']}\n"
    if state.get("previous_code"):
        user_msg += f"\nPrevious code:\n{state['previous_code']}\n"
    if state.get("feedback"):
        user_msg += f"\nReviewer feedback to address:\n{state['feedback']}\n"

    response = llm.invoke(
        [
            {"role": "system", "content": CODER_CONFIG.system_prompt},
            {"role": "user", "content": user_msg},
        ]
    )
    state["code"] = response.content
    return state


def build_coder_graph():
    graph = StateGraph(CoderState)
    graph.add_node("generate", _generate)
    graph.set_entry_point("generate")
    graph.add_edge("generate", END)
    return graph.compile()


coder_graph = build_coder_graph()

import re
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
from shared.config import REVIEWER_CONFIG
from shared.llm import get_llm


class ReviewerState(TypedDict):
    spec: str
    code: str
    verdict: str
    feedback: str


def _review(state: ReviewerState) -> ReviewerState:
    llm = get_llm(model=REVIEWER_CONFIG.model, temperature=REVIEWER_CONFIG.temperature)
    user_msg = f"Spec:\n{state['spec']}\n\nCode to review:\n{state['code']}\n"
    response = llm.invoke(
        [
            {"role": "system", "content": REVIEWER_CONFIG.system_prompt},
            {"role": "user", "content": user_msg},
        ]
    )
    text = response.content
    match = re.search(r"VERDICT:\s*(APPROVED|CHANGES_REQUESTED)", text, re.IGNORECASE)
    verdict = match.group(1).upper() if match else "CHANGES_REQUESTED"
    feedback = text[match.end():].strip() if match else text.strip()

    state["verdict"] = verdict
    state["feedback"] = feedback
    return state


def build_reviewer_graph():
    graph = StateGraph(ReviewerState)
    graph.add_node("review", _review)
    graph.set_entry_point("review")
    graph.add_edge("review", END)
    return graph.compile()


reviewer_graph = build_reviewer_graph()

Wrapping a graph as an A2A executor

For Wrapping a graph as an A2A executor, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness. For Wrapping a graph as an A2A executor, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks.task_updater import TaskUpdater
from a2a.types import Part, Task, TaskState, TaskStatus

from shared.coder_graph import coder_graph

class CoderAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        user_message = context.message
        task_id = context.task_id
        context_id = context.context_id

        await event_queue.enqueue_event(
            Task(
                id=task_id,
                context_id=context_id,
                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
                history=[user_message] if user_message else [],
            )
        )
        updater = TaskUpdater(
            event_queue=event_queue, task_id=task_id, context_id=context_id
        )
        working_message = updater.new_agent_message(
            parts=[Part(text="Generating code...")]
        )
        await updater.start_work(message=working_message)

        query = context.get_user_input()
        spec, previous_code, feedback = _parse_input(query)

        try:
            result = coder_graph.invoke(
                {
                    "spec": spec,
                    "previous_code": previous_code,
                    "feedback": feedback,
                    "code": "",
                }
            )
        except Exception as exc:
            error_message = updater.new_agent_message(
                parts=[Part(text=f"❌ Code generation failed: {exc}")]
            )
            await updater.failed(message=error_message)
            return

        await updater.add_artifact(
            parts=[Part(text=result["code"])], name="code", last_chunk=True
        )
        await updater.complete()

...

Serving each agent

When working through Serving each agent, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

import uvicorn

from starlette.applications import Starlette
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

from coder_agent.agent_executor import CoderAgentExecutor

HOST = "127.0.0.1"
PORT = 9001
RPC_URL = "/a2a/jsonrpc"

def build_agent_card() -> AgentCard:
    skill = AgentSkill(
        id="write_code",
        name="Write / revise Python code",
        description="Writes a Python function from a spec, or revises "
                    "previous code given reviewer feedback.",
        tags=["coding", "python"],
        examples=["SPEC: write a function that reverses a linked list"],
        input_modes=["text"],
        output_modes=["text", "task-status"],
    )
    return AgentCard(
        name="Coder Agent",
        description="Generates and revises Python code based on a spec "
                    "and optional reviewer feedback.",
        version="0.1.0",
        capabilities=AgentCapabilities(streaming=True, push_notifications=False),
        default_input_modes=["text"],
        default_output_modes=["text", "task-status"],
        skills=[skill],
        supported_interfaces=[
            AgentInterface(
                protocol_binding="JSONRPC",
                protocol_version="1.0",
                url=f"http://{HOST}:{PORT}{RPC_URL}",
            ),
        ],
    )

def main():
    agent_card = build_agent_card()
    request_handler = DefaultRequestHandler(
        agent_executor=CoderAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=agent_card,
    )

    routes = []
    routes.extend(create_agent_card_routes(agent_card=agent_card))
    routes.extend(create_jsonrpc_routes(request_handler=request_handler, rpc_url=RPC_URL))

    app = Starlette(routes=routes)
    uvicorn.run(app, host=HOST, port=PORT)


if __name__ == "__main__":
    main()

The orchestrator

When working through The orchestrator, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

import asyncio
import sys
import uuid

import httpx
from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import get_artifact_text, get_message_text
from a2a.types import Message, Part, Role, SendMessageRequest
from shared.config import MAX_ROUNDS

CODER_URL = "http://127.0.0.1:9001"
REVIEWER_URL = "http://127.0.0.1:9002"


async def get_client_for(httpx_client: httpx.AsyncClient, base_url: str):
    resolver = A2ACardResolver(httpx_client, base_url)
    card = await resolver.get_agent_card()
    return await create_client(card, client_config=ClientConfig(httpx_client=httpx_client))

async def send_and_collect(client, text: str) -> str:
    message = Message(
        role=Role.ROLE_USER,
        message_id=str(uuid.uuid4()),
        parts=[Part(text=text)],
        context_id=str(uuid.uuid4()),
    )
    request = SendMessageRequest(message=message)

    result_text = ""
    async for event in client.send_message(request):
        if event.HasField("message"):
            result_text = get_message_text(event.message, delimiter="\n")
        elif event.HasField("artifact_update"):
            result_text += get_artifact_text(
                event.artifact_update.artifact, delimiter="\n"
            )
        elif event.HasField("status_update"):
            status = event.status_update.status
            if status.state == TaskState.TASK_STATE_FAILED:
                error_text = (
                    get_message_text(status.message, delimiter="\n")
                    if status.HasField("message")
                    else "Task failed with no error message."
                )
                raise RuntimeError(error_text)
    return result_text

async def run(spec: str) -> dict:
    """
    Runs the coder<->reviewer loop for `spec`. Returns a result dict:
        {"approved": bool, "rounds": int, "code": str, "review": str}
    or, if a task failed, {"approved": False, "error": str, "code": str | None}.
    """
    async with httpx.AsyncClient(timeout=120) as httpx_client:
        coder = await get_client_for(httpx_client, CODER_URL)
        reviewer = await get_client_for(httpx_client, REVIEWER_URL)

        code = None
        feedback = None
        review_text = ""

        try:
            for round_num in range(1, MAX_ROUNDS + 1):
                print(f"\n=== Round {round_num}: Coder ===")
                coder_prompt = f"SPEC: {spec}\n"
                if code:
                    coder_prompt += f"PREVIOUS_CODE: {code}\nFEEDBACK: {feedback}\n"
                code = await send_and_collect(coder, coder_prompt)
                print(code)

                print(f"\n=== Round {round_num}: Reviewer ===")
                reviewer_prompt = f"SPEC: {spec}\nCODE: {code}"
                review_text = await send_and_collect(reviewer, reviewer_prompt)
                print(review_text)

                if review_text.startswith("VERDICT: APPROVED"):
                    print("\n✅ Approved. Final code:\n")
                    print(code)
                    return {
                        "approved": True,
                        "rounds": round_num,
                        "code": code,
                        "review": review_text,
                    }

                feedback = review_text.split("\n", 1)[-1]

            print(f"\n⚠️ Max rounds ({MAX_ROUNDS}) reached without approval. Last code:\n")
            print(code)
            return {
                "approved": False,
                "rounds": MAX_ROUNDS,
                "code": code,
                "review": review_text,
            }
        except RuntimeError as exc:
            print(f"\n{exc}")
            return {"approved": False, "error": str(exc), "code": code}
        finally:
            await coder.close()
            await reviewer.close()

Running it

When working through Running it, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through Running it, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

pip install -r requirements.txt

# terminal 1
python -m coder_agent

# terminal 2
python -m reviewer_agent

# terminal 3
python orchestrator.py "write a function that checks if a string is a palindrome"

curl http://127.0.0.1:9001/.well-known/agent-card.json

An example of an actual run

An example of an actual run works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Things that broke and how to un-break them

Things that broke and how to un-break them works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

ModuleNotFoundError: No module named 'a2a.server.apps'

ModuleNotFoundError: No module named 'a2a.server.apps' works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts. ModuleNotFoundError: No module named 'a2a.server.apps' works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

ModuleNotFoundError: No module named 'sse_starlette'

For ModuleNotFoundError: No module named 'sse_starlette', define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Mysterious timeouts around 5 seconds

For Mysterious timeouts around 5 seconds, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Reasoning models eat the round

For Reasoning models eat the round, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Prefer structured outputs with schema validation over free-form prose when the next step is code or a tool call. For Reasoning models eat the round, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

Protobuf enum names

When working through Protobuf enum names, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Part no longer wraps TextPart

When working through Part no longer wraps TextPart, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node.

Room to extend

When working through Room to extend, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Checkpoint after expensive steps. Resume should not re-bill the same LLM call when an operator retries a later node. When working through Room to extend, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

References

References works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Keep graph state flat and typed. Nested blobs hide which node wrote which field and break resume after interrupts.

Operational checklist

For Operational checklist, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state.

Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments.

Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Write a short runbook: how to rotate keys, how to drain the queue, how to roll back the last ingest.

Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.

Before promoting the stack, freeze versions, capture a golden transcript for the critical path, and confirm rollback steps. Shared environments need rate limits, tenancy checks, and a clear owner for secret rotation. Prefer boring reliability over clever one-off demos.

Batch note for 684455491963: keep provider keys out of the repo, set a per-session token ceiling, and store transcripts next to the eval fixtures so later model swaps stay comparable.

When working through hardening note 0, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.

Hardening detail 0/640: measure wall time, error class, and token spend for this note, then decide whether to keep the change based on a fixed question set rather than anecdote.

hardening note 1 works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline.

Hardening detail 1/640: measure wall time, error class, and token spend for this note, then decide whether to keep the change based on a fixed question set rather than anecdote.

For hardening note 2, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments.

Hardening detail 2/640: measure wall time, error class, and token spend for this note, then decide whether to keep the change based on a fixed question set rather than anecdote.