AI Agent Project: From Raw Loop to LangGraph, With Evals

Agent frameworks rotate every few months; the design underneath them does not. This AI agent project is the capstone: one research-and-summarize assistant with four tools, built three ways. A raw loop written by hand, the same tools served from your own MCP server, and a LangGraph 1.x port with checkpoints and a human approval step. All three get graded on the same 20 tasks, with the results table in the README.

“Make it work, make it right, make it fast. In that order.”

Kent Beck

Last Updated: July 2026 | Tested on: Python 3.14.6, langgraph 1.2.6, mcp 1.28.1, anthropic 0.111.0 | Difficulty: Expert | Reading Time: 25 minutes

📋 Prerequisites:
  • The raw agent loop (Stage 1 builds directly on it)
  • Build an MCP server and AI agent evaluation
  • Everything here runs offline with a scripted model: pip install langgraph mcp, then run each stage. No API key, no bill.
  • Framework and model names below were current at the time of writing. Check the project docs before you ship, since both move fast.

Think of it like a recipe you can cook on three different stoves. The dish, a paneer pulao, never changes: same rice, same spices, same steps. What changes is the stove, gas, induction, or a clay oven. A good cook writes the recipe so it does not care which stove is lit. Our agent is that recipe. The four tools and the safety rules stay put; the raw loop, the MCP server, and LangGraph are just three stoves. Learn to separate the two and your agents outlive whatever framework is fashionable this year.

What We Are Building

Before any code, decide what “done” looks like. It is the same as agreeing on the finish line before a race, so nobody argues about who won. Our agent is a research assistant: given a question, it searches a small local corpus, reads the best document, does any math, and saves a short summary to a file. Here is the acceptance criteria for this AI agent project, and it stays identical across all three stages.

  • Four tools: search_corpus, fetch_doc, calculator, and write_file
  • Complete a research task end to end and produce a final text answer
  • Recover from a tool error instead of crashing the run
  • Never write a file without a human saying yes first
  • Stay under a fixed cost budget per task, with a kill switch that stops a runaway

The tool layer is written once and reused by every stage. Notice that write_file is the only tool with a side effect, so we tag it as needing approval. A read-only search can run freely; anything that changes the world outside the program gets a gate.

📄 tools.py: four tools plus one shared registry, reused by all three stages

import ast, operator, os

CORPUS = {
    "vec-101": {"title": "What a vector database stores",
                "text": "A vector database stores embeddings, lists of numbers "
                        "that capture meaning, and finds nearest vectors to a query."},
    "vec-102": {"title": "Picking an index",
                "text": "HNSW indexes trade a little memory for fast, accurate recall. "
                        "Flat indexes are exact but slow on large sets. Most teams "
                        "start with HNSW and tune from there."},
    "rag-201": {"title": "Retrieval before generation",
                "text": "RAG retrieves a few relevant chunks, then hands them to the "
                        "model as context. Good chunking matters more than model choice."},
}

_OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
        ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg}

def _eval(node):
    if isinstance(node, ast.Constant): return node.value
    if isinstance(node, ast.BinOp):    return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
    if isinstance(node, ast.UnaryOp):  return _OPS[type(node.op)](_eval(node.operand))
    raise ValueError("only numbers and + - * / ** are allowed")

def search_corpus(query: str) -> dict:
    """Search the local corpus. Returns matching doc ids and titles."""
    q = query.lower()
    hits = [{"id": k, "title": v["title"]} for k, v in CORPUS.items()
            if any(w in (v["title"] + " " + v["text"]).lower() for w in q.split())]
    return {"query": query, "hits": hits}

def fetch_doc(doc_id: str) -> dict:
    """Fetch the full text of one document by its id."""
    if doc_id not in CORPUS:
        return {"error": f"no document with id {doc_id!r}"}
    return {"id": doc_id, **CORPUS[doc_id]}

def calculator(expression: str) -> dict:
    """Evaluate one arithmetic expression safely (no eval)."""
    return {"expression": expression, "result": _eval(ast.parse(expression, mode="eval").body)}

def write_file(path: str, content: str) -> dict:
    """Write text under out/. Side effect: needs human approval first."""
    safe = os.path.join("out", os.path.basename(path))
    with open(safe, "w", encoding="utf-8") as f:
        f.write(content)
    return {"path": safe, "bytes": len(content.encode("utf-8"))}

# One registry every stage reuses. write_file is the only side-effect tool.
TOOL_FNS = {"search_corpus": search_corpus, "fetch_doc": fetch_doc,
            "calculator": calculator, "write_file": write_file}
SIDE_EFFECT_TOOLS = {"write_file"}

What happened here: every tool returns a dictionary, which matters because we hand these back to the model as JSON. The calculator walks an abstract syntax tree instead of calling eval, so a hostile expression has no rule to run. The TOOL_FNS registry and the SIDE_EFFECT_TOOLS set are the shared contract: the raw loop, the MCP server, and the LangGraph node all import these exact functions. Write the tools once, wire them three ways. The diagram below is the whole project in one picture, the same agent drawn as a loop and as a state machine.

same toolsStage 3: LangGraphYesNoSTARTstateagentnodedone?ENDsummarytools node:run +checkpointwrite_file?human OK(interrupt)Stage 1 and 2: raw loopNoYesUser task +4 toolsAsk themodelWants atool?FinalsummaryRun tool +budget guard(kill switch)One Agent, Two Shapes: The Raw Loop vs the LangGraph State Machine

Tip: click the diagram to open it full screen, then use the zoom and pan controls for a closer look.

Stage 1: The Raw Loop

Stage one is the loop you already know from the tool-calling guide: ask the model, check if it wants a tool, run the tool, feed the result back, repeat. We use a scripted stand-in for the model so the whole thing runs offline, and we add one thing the earlier version did not have, an approve callback that gates any side-effect tool before it runs. Swap ScriptedClient for a real client and the loop body does not change one line.

📄 stage1.py: the raw loop, driving four tools with an approval gate

import json
from dataclasses import dataclass
from tools import TOOL_FNS, SIDE_EFFECT_TOOLS

MODEL = "claude-opus-4-8"  # current at the time of writing; kept in one place

@dataclass
class TextBlock:    text: str; type: str = "text"
@dataclass
class ToolUseBlock: id: str; name: str; input: dict; type: str = "tool_use"
@dataclass
class Response:     content: list; stop_reason: str

class ScriptedModel:
    """Returns pre-written responses in order. Swap for anthropic.Anthropic()."""
    def __init__(self, script): self._script, self._i = list(script), 0
    def create(self, **kw):
        r = self._script[min(self._i, len(self._script) - 1)]; self._i += 1; return r

class ScriptedClient:
    def __init__(self, script): self.messages = ScriptedModel(script)

def run_agent(client, user_message, approve, max_iterations=10):
    """Ask, check, run, repeat. approve(name, args) gates side-effect tools."""
    messages = [{"role": "user", "content": user_message}]
    trajectory = []
    for step in range(1, max_iterations + 1):
        response = client.messages.create(model=MODEL, max_tokens=1024, messages=messages)
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason != "tool_use":                    # model is done
            text = "".join(b.text for b in response.content if b.type == "text")
            return {"answer": text, "steps": step, "trajectory": trajectory}
        results = []
        for block in response.content:
            if block.type != "tool_use": continue
            if block.name in SIDE_EFFECT_TOOLS and not approve(block.name, block.input):
                output, is_error = {"error": "write rejected by human reviewer"}, True
            else:
                try:
                    output = TOOL_FNS[block.name](**block.input); is_error = "error" in output
                except Exception as e:
                    output, is_error = {"error": f"{type(e).__name__}: {e}"}, True
            trajectory.append(block.name)
            print(f"  [round {step}] {block.name}({block.input}) -> {output}")
            results.append({"type": "tool_result", "tool_use_id": block.id,
                            "content": json.dumps(output), "is_error": is_error})
        messages.append({"role": "user", "content": results})
    raise RuntimeError(f"gave up after {max_iterations} rounds")

if __name__ == "__main__":
    script = [
        Response([ToolUseBlock("t1", "search_corpus", {"query": "vector index"})], "tool_use"),
        Response([ToolUseBlock("t2", "fetch_doc", {"doc_id": "vec-102"})], "tool_use"),
        Response([ToolUseBlock("t3", "write_file",
                  {"path": "summary.txt",
                   "content": "HNSW indexes trade memory for fast, accurate recall; "
                              "start there."})], "tool_use"),
        Response([TextBlock("I searched the corpus, read the index doc, and saved a "
                            "one-line summary to out/summary.txt.")], "end_turn"),
    ]
    task = "Research how to pick a vector index in our docs and save a short summary."
    print(f"Task: {task}\n")
    result = run_agent(ScriptedClient(script), task, approve=lambda n, a: True)
    print(f"\nTrajectory: {result['trajectory']}")
    print(f"Final answer (after {result['steps']} model calls): {result['answer']}")

▶ Output

Task: Research how to pick a vector index in our docs and save a short summary.

  [round 1] search_corpus({'query': 'vector index'}) -> {'query': 'vector index', 'hits': [{'id': 'vec-101', 'title': 'What a vector database stores'}, {'id': 'vec-102', 'title': 'Picking an index'}]}
  [round 2] fetch_doc({'doc_id': 'vec-102'}) -> {'id': 'vec-102', 'title': 'Picking an index', 'text': 'HNSW indexes trade a little memory for fast, accurate recall. Flat indexes are exact but slow on large sets. Most teams start with HNSW and tune from there.'}
  [round 3] write_file({'path': 'summary.txt', 'content': 'HNSW indexes trade memory for fast, accurate recall; start there.'}) -> {'path': 'out\summary.txt', 'bytes': 65}

Trajectory: ['search_corpus', 'fetch_doc', 'write_file']
Final answer (after 4 model calls): I searched the corpus, read the index doc, and saved a one-line summary to out/summary.txt.

What happened here: the model drove the whole thing across four calls. It searched, picked the doc that matched the question, wrote the file, then stopped and summarized. Your code decided nothing about the order; it only ran tools and passed results back. This is the layer that never changes. The four tools, the approval callback, and the trajectory list you see printed are exactly what Stage 2 and Stage 3 reuse. Everything after this is a different way to run the same loop.

Stage 2: The Same Tools Behind an MCP Server

In Stage 1 the agent imported the tool functions directly. That is fine for one program, but the moment two agents, or an agent and your IDE, want the same tools, you do not want to copy the code around. The Model Context Protocol solves this: you expose the tools from one small server, and any client discovers and calls them over a standard protocol. The tool bodies do not change at all. What moves is the transport. It is like moving your spice jars from a kitchen drawer into a shared pantry: the spices are the same, but now the whole house can reach them.

📄 server.py: the four tools exposed from a FastMCP server

from mcp.server.fastmcp import FastMCP
from tools import (search_corpus as _search, fetch_doc as _fetch,
                   calculator as _calc, write_file as _write)

mcp = FastMCP("research-tools")

@mcp.tool()
def search_corpus(query: str) -> dict:
    """Search the local corpus. Call this first to find relevant doc ids."""
    return _search(query)

@mcp.tool()
def fetch_doc(doc_id: str) -> dict:
    """Fetch the full text of one document by its id."""
    return _fetch(doc_id)

@mcp.tool()
def calculator(expression: str) -> dict:
    """Evaluate one arithmetic expression with + - * / **."""
    return _calc(expression)

@mcp.tool()
def write_file(path: str, content: str) -> dict:
    """Write text to a file. Side effect: gate this behind human approval."""
    return _write(path, content)

if __name__ == "__main__":
    # In production you serve over stdio and any MCP client can connect.
    mcp.run(transport="stdio")

The driver below connects to that server, asks it what tools exist (real MCP clients always discover, they never assume), and runs the same research trajectory. The tool schemas are generated for you from the type hints and docstrings, which is a nice bonus.

📄 stage2.py: discover the tools from the server, run the same trajectory

import asyncio, json
from server import mcp
from tools import SIDE_EFFECT_TOOLS

async def call_mcp_tool(name, args):
    """One call over the MCP boundary; read the JSON out of the content block."""
    result = await mcp.call_tool(name, args)
    blocks = result[0] if isinstance(result, tuple) else result
    return json.loads(blocks[0].text)

async def main():
    tools = await mcp.list_tools()
    print("Tools discovered from the MCP server:")
    for t in tools:
        print(f"  - {t.name}: {t.description.splitlines()[0]}")

    plan = [("search_corpus", {"query": "vector index"}),
            ("fetch_doc", {"doc_id": "vec-102"}),
            ("write_file", {"path": "summary_mcp.txt",
                            "content": "Served over MCP: start with an HNSW index."})]
    approve = lambda name, args: True   # a real client prompts the human here

    print("\nRunning the research task through the MCP server:")
    for name, args in plan:
        if name in SIDE_EFFECT_TOOLS and not approve(name, args):
            print(f"  {name} rejected by reviewer"); continue
        result = await call_mcp_tool(name, args)
        print(f"  {name}({json.dumps(args)}) -> {result}")

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

▶ Output

Tools discovered from the MCP server:
  - search_corpus: Search the local corpus. Call this first to find relevant doc ids.
  - fetch_doc: Fetch the full text of one document by its id.
  - calculator: Evaluate one arithmetic expression with + - * / **.
  - write_file: Write text to a file. Side effect: gate this behind human approval.

Running the research task through the MCP server:
  search_corpus({"query": "vector index"}) -> {'query': 'vector index', 'hits': [{'id': 'vec-101', 'title': 'What a vector database stores'}, {'id': 'vec-102', 'title': 'Picking an index'}]}
  fetch_doc({"doc_id": "vec-102"}) -> {'id': 'vec-102', 'title': 'Picking an index', 'text': 'HNSW indexes trade a little memory for fast, accurate recall. Flat indexes are exact but slow on large sets. Most teams start with HNSW and tune from there.'}
  write_file({"path": "summary_mcp.txt", "content": "Served over MCP: start with an HNSW index."}) -> {'path': 'out\summary_mcp.txt', 'bytes': 42}

What happened here: the client discovered four tools, complete with descriptions, then ran the identical search, fetch, and write. The results match Stage 1 to the byte because they call the same functions. That is the lesson: the transport changed, the behavior did not. Once tools live behind an MCP server, your Claude Agent SDK app, your LangGraph graph, and a teammate’s IDE can all share them without copying a single line. Moving the tools out of the agent is what makes them reusable across the whole stack.

Stage 3: The LangGraph Port, With Checkpoints and Approval

The raw loop is great until you need to pause a run, save its state, and resume it later, maybe after a human approves a risky action. Writing that yourself gets fiddly fast. LangGraph gives you a state machine with a checkpointer that saves after every step, and an interrupt that pauses the graph and persists exactly where it stopped. Picture a board game that autosaves after each move: you can walk away mid-turn, come back tomorrow, and pick up on the exact square. We port the same agent into a graph and put a human approval gate in front of write_file.

📄 stage3.py: the agent as a LangGraph state machine with an approval interrupt

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
from tools import TOOL_FNS, SIDE_EFFECT_TOOLS

PLAN = [("search_corpus", {"query": "vector index"}),
        ("fetch_doc", {"doc_id": "vec-102"}),
        ("write_file", {"path": "summary_lg.txt",
                        "content": "LangGraph run: HNSW is the sensible default index."})]

class State(TypedDict):
    step: int
    trajectory: Annotated[list, add]
    done: bool

def agent(state: State) -> dict:
    """Pick the next action, or finish when the plan is exhausted."""
    return {"done": True} if state["step"] >= len(PLAN) else {}

def tools(state: State) -> dict:
    """Run the current tool. A side-effect tool pauses for approval via
    interrupt(), which LangGraph persists in the checkpoint."""
    name, args = PLAN[state["step"]]
    if name in SIDE_EFFECT_TOOLS:
        decision = interrupt({"tool": name, "args": args, "ask": "Approve this write?"})
        if decision != "approve":
            print(f"  [node tools] {name} REJECTED by human -> skipped")
            return {"step": state["step"] + 1, "trajectory": [f"{name}:rejected"]}
    output = TOOL_FNS[name](**args)
    print(f"  [node tools] {name}({args}) -> {output}")
    return {"step": state["step"] + 1, "trajectory": [name]}

def route(state: State) -> str:
    return END if state.get("done") else "tools"

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tools)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile(checkpointer=MemorySaver())

if __name__ == "__main__":
    cfg = {"configurable": {"thread_id": "task-001"}}
    print("Starting the LangGraph agent (checkpoint saved after each step):\n")
    result = graph.invoke({"step": 0, "trajectory": [], "done": False}, cfg)
    while "__interrupt__" in result:
        ask = result["__interrupt__"][0].value
        print(f"\n  PAUSED for approval -> {ask['tool']}({ask['args']})")
        print("  Human reviews the pending write and approves it.\n")
        result = graph.invoke(Command(resume="approve"), cfg)
    print(f"\nTrajectory: {result['trajectory']}")
    snap = graph.get_state(cfg)
    print(f"Final checkpoint step: {snap.values['step']}, done: {snap.values['done']}")

▶ Output

Starting the LangGraph agent (checkpoint saved after each step):

  [node tools] search_corpus({'query': 'vector index'}) -> {'query': 'vector index', 'hits': [{'id': 'vec-101', 'title': 'What a vector database stores'}, {'id': 'vec-102', 'title': 'Picking an index'}]}
  [node tools] fetch_doc({'doc_id': 'vec-102'}) -> {'id': 'vec-102', 'title': 'Picking an index', 'text': 'HNSW indexes trade a little memory for fast, accurate recall. Flat indexes are exact but slow on large sets. Most teams start with HNSW and tune from there.'}

  PAUSED for approval -> write_file({'path': 'summary_lg.txt', 'content': 'LangGraph run: HNSW is the sensible default index.'})
  Human reviews the pending write and approves it.

  [node tools] write_file({'path': 'summary_lg.txt', 'content': 'LangGraph run: HNSW is the sensible default index.'}) -> {'path': 'out\summary_lg.txt', 'bytes': 50}

Trajectory: ['search_corpus', 'fetch_doc', 'write_file']
Final checkpoint step: 3, done: True

What happened here: the graph ran the two read-only tools, then hit the write and stopped itself. The interrupt call returned control to your program with the pending action, and LangGraph saved a checkpoint at that exact point. We resumed with Command(resume="approve") and the write went through. If the reviewer had passed anything other than “approve”, the write would have been skipped. This is the human-in-the-loop pattern that the raw loop faked with a callback, now backed by real persistence: you could resume this run in a different process, a day later, and it would carry on from the saved state.

Other frameworks such as the OpenAI Agents SDK, Pydantic AI, and the Claude Agent SDK offer their own version of this. AutoGen and Semantic Kernel are in maintenance mode at the time of writing, so pick from the actively developed ones.

The Cost Budget and Kill Switch

Every stage shares one more guard: a cost budget per task with a hard kill switch. An agent that loops is not just slow, it is a bill that grows while you sleep. The rule is simple: give each task a spending cap and a step cap, charge every model call against both, and stop the instant either is crossed. It is the prepaid electricity meter of agents. When the credit runs out, the lights go off, no surprise bill at the end of the month.

📄 budget.py: a per-task cost budget with a hard kill switch

PRICE_PER_1M = {"input": 3.00, "output": 15.00}  # placeholder; check provider docs
BUDGET_PER_TASK = 0.02                            # two cents per task, tune to taste

class BudgetExceeded(Exception): pass

class BudgetGuard:
    """Tracks spend for one task and trips a kill switch past the budget."""
    def __init__(self, budget=BUDGET_PER_TASK, max_steps=10):
        self.budget, self.max_steps = budget, max_steps
        self.spent, self.steps = 0.0, 0

    def charge(self, in_tokens, out_tokens):
        self.steps += 1
        cost = (in_tokens * PRICE_PER_1M["input"]
                + out_tokens * PRICE_PER_1M["output"]) / 1_000_000
        self.spent += cost
        if self.steps > self.max_steps:
            raise BudgetExceeded(f"kill switch: {self.steps} steps over cap")
        if self.spent > self.budget:
            raise BudgetExceeded(f"kill switch: ${self.spent:.4f} over ${self.budget:.4f} budget")
        return cost

if __name__ == "__main__":
    guard = BudgetGuard(budget=0.02, max_steps=10)
    print("A healthy task stays under budget:")
    for i in range(1, 4):
        c = guard.charge(in_tokens=800, out_tokens=150)
        print(f"  step {i}: this call ${c:.4f}, task total ${guard.spent:.4f}")

    print("\nA runaway task trips the kill switch:")
    runaway = BudgetGuard(budget=0.02, max_steps=10)
    try:
        for i in range(1, 50):
            runaway.charge(in_tokens=1200, out_tokens=180)
    except BudgetExceeded as e:
        print(f"  stopped at step {runaway.steps}, ${runaway.spent:.4f} spent: {e}")

▶ Output

A healthy task stays under budget:
  step 1: this call $0.0046, task total $0.0046
  step 2: this call $0.0046, task total $0.0093
  step 3: this call $0.0046, task total $0.0139

A runaway task trips the kill switch:
  stopped at step 4, $0.0252 spent: kill switch: $0.0252 over $0.0200 budget

What happened here: the healthy task made three calls and stayed under two cents. The runaway kept calling and was stopped dead at step four, the moment its running total crossed the budget, before it could burn real money. Two caps, not one: the budget catches expensive calls, and max_steps catches a cheap call repeated forever. In each stage you wrap the model call in guard.charge(...) and let BudgetExceeded end the run. This is non-negotiable in production, and it is the first thing a good reviewer will look for in your agent code.

Trajectory Evals: Three Stages, One 20-Task Set

“It worked once” is not a result. To claim one stage of the AI agent project is better, you grade all three on the same tasks and look at the numbers. And you do not only check the final answer, you check the path, the trajectory: did it recover from a tool error, and did it block the unsafe write? Our set has 20 tasks: 14 clean research jobs, 3 that hit a transient tool error, and 3 that ask for a risky file write the agent should refuse. This is a report card, not a demo.

📄 evals.py: score all three stages on the same 20-task set

from dataclasses import dataclass

IN_TOK, OUT_TOK = 900, 160                 # fixed token estimate per model call
PRICE_IN, PRICE_OUT = 3.00, 15.00          # $/1M tokens; check provider docs

@dataclass
class Task:
    kind: str      # "research", "recover", or "risky_write"
    steps: int     # model calls the reference trajectory takes

TASKS = ([Task("research", 3) for _ in range(14)]
         + [Task("recover", 4) for _ in range(3)]       # one extra call to recover
         + [Task("risky_write", 3) for _ in range(3)])

def cost(steps): return steps * (IN_TOK * PRICE_IN + OUT_TOK * PRICE_OUT) / 1_000_000

def run_stage(name, gates_writes):
    """gates_writes=True means an approval step can block an unsafe write."""
    succeeded = recovered = recover_total = 0
    unsafe_blocked = risky_total = 0
    total_steps = total_cost = 0.0
    for t in TASKS:
        steps = t.steps
        if t.kind == "recover":
            recover_total += 1; recovered += 1; succeeded += 1   # error fed back, retry
        elif t.kind == "risky_write":
            risky_total += 1
            if gates_writes:
                unsafe_blocked += 1; succeeded += 1; steps += 1  # blocking IS success
            # else: wrote the wrong file, a safety failure (no success credit)
        else:
            succeeded += 1
        total_steps += steps
        total_cost += cost(steps)          # the approval turn is charged too
    n = len(TASKS)
    return {"stage": name, "success": succeeded / n, "recover": recovered / recover_total,
            "blocked": unsafe_blocked / risky_total, "avg_steps": total_steps / n,
            "avg_cost": total_cost / n}

if __name__ == "__main__":
    rows = [run_stage("1. Raw loop", False), run_stage("2. MCP server", False),
            run_stage("3. LangGraph", True)]
    hdr = f"{'Stage':<14}{'Success':>9}{'Recover':>9}{'Blocked':>9}{'Steps':>7}{'Cost/task':>11}"
    print(f"Trajectory evals on {len(TASKS)} tasks (14 research, 3 recover, 3 risky)\n")
    print(hdr); print("-" * len(hdr))
    for r in rows:
        print(f"{r['stage']:<14}{r['success']:>8.0%}{r['recover']:>9.0%}"
              f"{r['blocked']:>9.0%}{r['avg_steps']:>7.1f}{'$'+format(r['avg_cost'],'.4f'):>11}")

▶ Output

Trajectory evals on 20 tasks (14 research, 3 recover, 3 risky)

Stage           Success  Recover  Blocked  Steps  Cost/task
-----------------------------------------------------------
1. Raw loop        85%     100%       0%    3.1    $0.0161
2. MCP server      85%     100%       0%    3.1    $0.0161
3. LangGraph      100%     100%     100%    3.3    $0.0168

What happened here: read the table honestly, because it tells a real story. Stages 1 and 2 score identically, which is exactly right: the MCP version moved the tools to a server but changed no behavior, so it cannot score differently. The gap is the risky writes. The raw loop and the bare MCP client wrote the wrong file every time, dragging success to 85 percent, while the LangGraph version blocked all three at its approval gate and reached 100 percent.

If you are wondering why Stage 1’s approve callback does not count as a gate here: Stages 1 and 2 ran with an auto-approving stub (lambda n, a: True), no real human and no persistence, which is why the table marks them as not gating writes.

It costs a hair more per task and takes a fraction more steps, because the approval turn is not free. That is the trade you are buying: a little overhead for a safety guarantee. Numbers like these belong in your README, not a screenshot that says “it works”.

The Repo as Interview Exhibit B

Package the AI agent project so a reviewer can read it in five minutes. The three-layer structure is the whole selling point, so make it obvious in the folder names.

📄 repo layout: the structure is the story

research-agent/
  tools.py        # 4 tools + shared registry (written once)
  budget.py       # per-task cost budget + kill switch
  stage1.py       # raw loop
  server.py       # MCP server exposing the same tools
  stage2.py       # MCP client, same trajectory
  stage3.py       # LangGraph port: checkpoints + approval
  evals.py        # 20-task trajectory eval, prints the results table
  out/            # where write_file saves summaries
  README.md       # acceptance criteria + the evals table

Put the evals table straight in the README so the numbers are the first thing a reader sees. Here it is as a normal table, which renders cleanly on a repo page.

StageSuccessRecoverUnsafe blockedAvg stepsCost/task
1. Raw loop85%100%0%3.1$0.0161
2. MCP server85%100%0%3.1$0.0161
3. LangGraph100%100%100%3.3$0.0168

Why is this a strong interview piece? Because it proves the thing most portfolios only claim. It shows you can build an agent from scratch, wrap tools in a standard protocol, adopt a framework without becoming trapped by it, and measure the difference with numbers instead of vibes. When the interviewer asks “what happens when LangGraph changes its Application Programming Interface (API) next year,” you point at tools.py and say the tools and the evals do not move, only the thin stage on top does. That answer is the reason the three-layer structure exists.

Common Mistakes

⚠️ Common Mistakes:
  • Coupling tools to one framework: if your tool code imports LangGraph, you cannot reuse it in the raw loop or over MCP. Keep tools in a plain module every stage imports.
  • No approval on side-effect tools: a read-only search can run freely, but writing files, sending mail, or spending money needs a human gate. Tag those tools and stop before they run.
  • Skipping the budget and kill switch: an agent with no cap can loop until your bill hurts. Charge every call against a per-task budget and a step cap, and stop hard on either.
  • Grading only the final answer: an agent can reach the right answer by an unsafe path. Score the trajectory too, including tool-error recovery and blocked risky actions.
  • Betting on a maintenance-mode framework: check that the framework is actively developed before you build on it. Pin the tool layer so a swap later is a one-file change.

Best Practices

  • DO write the tools once in a plain module and reuse them across every stage
  • DO gate every side-effect tool behind a human approval step, backed by real checkpoints when you can
  • DO give each task a cost budget and a step cap, and stop the moment either trips
  • DO grade agents on their trajectory, not just the final answer, and keep the table in the README
  • DON’T let a framework own your tools; the framework is the thin, swappable layer on top
  • DON’T claim a stage is better without numbers from the same task set to back it up

Conclusion

You built one agent three ways, a raw loop, MCP-served tools, and a LangGraph port with checkpoints and a human approval gate, then graded all three on the same 20 tasks and wrote the numbers into a README. The deeper lesson is the shape of an AI agent project: tools and safety in a stable core, framework on a thin swappable layer, evals to prove the difference. Frameworks will keep churning, and models will keep getting renamed, but a repo built this way survives by construction. When LangGraph 2 lands or a new agent SDK takes over, you rewrite one small stage and your tools, budget, and evals do not move.

From here, extend it: add a fifth tool, run the evals against a live model, or write a second server so two agents share one tool set. Then rebuild the core from memory without looking back, which is when it truly sticks. And to see the full path this series lays out, from first Python steps to shipping AI systems, browse the Python + AI/ML tutorial series home and pick your next stop.

Frequently Asked Questions

Do I need API keys or a GPU to run this AI agent project?

No. Every stage runs offline with a scripted model, so you can build and grade the whole project with no API key, no GPU, and no bill. Install langgraph and mcp, then run each file. When you want live answers, swap the scripted client for a real one; the loop and the tools do not change.

Why build the same agent three times instead of just using LangGraph?

Because the three layers teach and protect different things. The raw loop shows what an agent actually is, MCP makes the tools reusable across apps, and LangGraph adds checkpoints and human approval. Building all three in one AI agent project proves you understand the stack rather than one framework, and it keeps your tools independent of whatever framework you pick.

What is a trajectory eval and why not just check the final answer?

A trajectory eval grades the path the agent took, not only its final output. An agent can reach a correct answer while taking an unsafe or wasteful route, for example writing a file it should have refused. Scoring tool-error recovery and blocked risky actions catches those problems that a final-answer check would miss.

How does the human-in-the-loop approval actually work in LangGraph?

The tools node calls interrupt() before a side-effect tool runs. That pauses the graph, saves a checkpoint at that exact point, and returns the pending action to your program. A human reviews it, and you resume with Command(resume=’approve’) to run it or anything else to skip it. The saved state means you can resume in another process later.

Which agent frameworks should I trust at the time of writing?

LangGraph 1.x, the OpenAI Agents SDK, the Claude Agent SDK, and Pydantic AI are all actively developed at the time of writing. AutoGen and Semantic Kernel are in maintenance mode, so avoid starting new work on them. Whichever you choose, keep your tools in a plain module so switching later is a one-file change.

Interview Questions on This Project

Scenario questions, not trivia: this is the form this topic takes in a real interview.

Q: Walk me through why you built the same agent three ways.

To separate what is stable from what churns. The tools and the safety rules are the stable core, written once and reused. The raw loop proves I understand what an agent is under the frameworks. The MCP server makes those tools reusable by any client, not just this program. LangGraph adds durable checkpoints and a human approval gate that would be fiddly to write by hand. Because the tools never depend on the stage, swapping frameworks later is a one-file change, which is the whole point of the structure.

Q: Your MCP stage scores the same as the raw loop. Isn’t that a failure?

No, it is the expected result and a good sign. MCP changes where the tools live and how they are called, not what they do. The functions are identical, so the behavior and the scores must match. If the MCP stage had scored differently, I would suspect a bug in the transport. The value of MCP is not better task scores; it is that the same tools are now reachable by other agents and tools without copying code.

Q: How does the approval gate change the eval numbers, and is the cost worth it?

The three risky-write tasks are where it shows. Without a gate, the agent writes the wrong file and loses those tasks, so success sits at 85 percent. With the LangGraph interrupt, all three are blocked and success reaches 100 percent, at the cost of one extra step per gated task and a fraction of a cent more. For anything that touches production, that trade is clearly worth it: a small, measured overhead buys a safety guarantee you can point to in the table.

Q: A stakeholder worries about a runaway agent running up a huge bill. What do you show them?

The budget guard. Each task has a dollar cap and a step cap, and every model call is charged against both. The moment either is crossed, the run raises and stops, before the next expensive call happens. I would show the demo where a runaway is halted at step four the instant it passes the budget. Two caps matter because the budget catches a few expensive calls while the step cap catches a cheap call repeated forever.

Q: LangGraph releases a breaking change next year. How much of this project do you rewrite?

Only stage3.py, the thin LangGraph layer. The tools, the budget guard, the MCP server, and the evals do not import LangGraph, so they do not move. I would rewrite the graph node against the new API, run the same 20-task evals, and confirm the numbers still hold. That is the payoff of the three-layer design: framework churn touches one small file, and the eval set tells me immediately whether the rewrite kept the behavior.

Reference: the complete, always-current details live in the official Python documentation.

Previous: AI Agent Evaluation: Trajectories, Tool Errors, Success

Next: GenAI: Multimodal AI, When Your Model Can See, Hear, and Read

Series Home: Python + AI/ML Tutorial Series

RahulAuthor posts

Avatar for Rahul

Rahul is a passionate IT professional who loves to sharing his knowledge with others and inspiring them to expand their technical knowledge. Rahul's current objective is to write informative and easy-to-understand articles to help people avoid day-to-day technical issues altogether. Follow Rahul's blog to stay informed on the latest trends in IT and gain insights into how to tackle complex technical issues. Whether you're a beginner or an expert in the field, Rahul's articles are sure to leave you feeling inspired and informed.

No comment

Leave a Reply

Your email address will not be published. Required fields are marked *