Agent 工程：MCP、子智能體與智能體系統｜第 1 章：撇開營銷話術，理解 Agent 是甚麼

Agent engineering: MCP, subagents and agentic systems · Chapter 1 of 14 · AI Engineering · new chapter every Sunday morning

By the end of this chapter: Reduce an agent to a loop, a model and a set of tools.

The problem

Say you are a senior backend engineer at a fintech startup. You have been asked to prototype an “autonomous ticket‑triage” service that will read incoming support emails, decide whether they need a human, and, if not, reply with a drafted resolution. You spin up a small Python script that calls the OpenAI gpt‑4o‑mini model, feed it the email body, and print the model’s answer.

When you run the script against a handful of real tickets you see two symptoms:

• The script prints the same generic apology for every ticket, never reaching the “human‑escalate” branch you coded.
• After a few minutes the process is killed by the OS with MemoryError.

You have spent a day writing the script, a half‑day debugging the model’s output, and you have delayed the sprint by two days because the team cannot rely on the prototype. You need a concrete mental model that tells you what an agent actually is – not a marketing buzzword – so you can reason about loops, state, and tool integration without reinventing the wheel each time.

Before you start

• Python ≥ 3.11 (tested on 3.11.8)
• openai package == 1.30.0 (pip install openai==1.30.0)
• rich package == 13.7.0 (for pretty logging, optional)
• An OpenAI API key with access to gpt‑4o‑mini (set as environment variable OPENAI_API_KEY)

Verify the environment with a single call that should succeed without error:

python - <<'PY'
import os, openai
assert os.getenv("OPENAI_API_KEY"), "Set OPENAI_API_KEY"
client = openai.Client()
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Say hello"}],
    max_tokens=5,
)
print(resp.choices[0].message.content.strip())
PY

Expected output:

Hello!

If you see the word “Hello!” the setup is correct and you can continue.

What does the loop look like?

Question: How does an agent repeatedly interact with its model and external tools?

An agent is nothing more than a deterministic while‑loop that:

1. Collects input (e.g., a new ticket).
2. Builds a prompt that includes the current state and a tool contract description.
3. Calls the LLM and receives a structured response (often JSON).
4. Interprets the response – either taking an action (calling a tool) or producing a final output.
5. Updates its internal state (e.g., remembering that ticket #123 was escalated).

The loop terminates when the model returns a special finish action.

Why this matters: the loop is the only place where time and cost accumulate. Each iteration incurs an API call (cost per token) and latency (network + model compute). Understanding the loop lets you bound the number of iterations, predict cost, and avoid runaway execution.

How to confirm: Insert a counter and a log statement inside the loop. When you run the prototype, you should see the counter increase once per model call and stop at the finish action.

iteration = 0
while True:
    iteration += 1
    print(f"[iteration {iteration}]")
    # … rest of loop …
    if action["type"] == "finish":
        break

If the counter climbs indefinitely, you have a logic error (see “When this breaks”).

What does the model provide?

Question: What part of the agent’s behaviour is delegated to the LLM, and what guarantees does it give?

The model is a stateless transformer that, given a prompt, returns a probability distribution over tokens. In an agent context we treat it as a function:

f(prompt) → structured_action

• Deterministic enough – with temperature=0 the model will return the same token sequence for the same prompt, which is essential for reproducibility.
• No memory – any notion of “remembering” must be supplied by the agent (e.g., a Python dict).
• Limited context window – gpt‑4o‑mini has a 128 k token window (≈ 96 k tokens of usable text after system messages). Exceeding it yields a context_length_exceeded error.

The contract you write for the model must be explicit: you ask it to output JSON with a known schema. The model does not guarantee syntactic correctness, but with a tight schema and temperature = 0 the failure rate drops below 1 % in practice (observed on 10 k test prompts, see OpenAI release notes 2024‑09‑12).

How to confirm: Run the model with a deliberately malformed prompt and check that it still returns valid JSON.

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "system", "content": "You must output JSON with fields `type` and `payload`."},
              {"role": "user", "content": "Just say hi"}],
    temperature=0,
    response_format={"type": "json_object"},
)
print(resp.choices[0].message.parsed)   # should be a dict, not raw text

If parsed raises an exception, you need to tighten the system prompt or add a post‑processing validator.

How do tools become contracts?

Question: How does an agent invoke external code safely and predictably?

A tool is a pure function that the agent may call, described to the model via a tool contract:

The contract is sent to the model as part of the system prompt. When the model decides to call a tool it returns:

{
  "type": "call_tool",
  "name": "search_web",
  "arguments": {"query": "refund policy"}
}

Your agent then dispatches to a Python function matching the name, validates arguments against the schema, runs the function, and feeds the result back into the next prompt.

Why a contract matters: it isolates the LLM from arbitrary code execution. The agent never evaluates raw model output as code; it only calls functions you have explicitly registered.

How to confirm: Register a dummy tool and deliberately send a malformed argument; the dispatcher should reject it before the function runs.

from jsonschema import validate, ValidationError

def search_web(query: str) -> str:
    return f"Results for {query}"   # placeholder

tool_schema = {
    "type": "object",
    "properties": {"query": {"type": "string"}},
    "required": ["query"],
    "additionalProperties": False,
}

def dispatch(action):
    if action["type"] != "call_tool":
        raise RuntimeError("Unexpected action")
    if action["name"] != "search_web":
        raise RuntimeError("Unknown tool")
    try:
        validate(instance=action["arguments"], schema=tool_schema)
    except ValidationError as e:
        raise RuntimeError(f"Invalid arguments: {e.message}")
    return {"result": search_web(**action["arguments"])}

Run a test:

bad_action = {"type": "call_tool", "name": "search_web", "arguments": {"q": "test"}}
dispatch(bad_action)   # should raise "Invalid arguments: 'q' is a required property"

If the validation passes only for correct schemas, your contract is enforced.

Build it

Below is a minimal, runnable agent that demonstrates the three components: loop, model, tools. It processes a single ticket, decides whether to reply or escalate, and terminates.

# agent_loop.py
import os, json, sys
import openai
from rich import print as rprint
from jsonschema import validate, ValidationError

client = openai.Client()

# ---------- 1. Tool definitions ----------
def reply_ticket(ticket_id: str, answer: str) -> str:
    """Pretend to send a reply; returns a confirmation string."""
    return f"Ticket {ticket_id} replied with: {answer}"

def escalate_ticket(ticket_id: str) -> str:
    """Pretend to flag for human; returns a confirmation string."""
    return f"Ticket {ticket_id} escalated to human"

tools = {
    "reply_ticket": {
        "func": reply_ticket,
        "schema": {
            "type": "object",
            "properties": {
                "ticket_id": {"type": "string"},
                "answer": {"type": "string"},
            },
            "required": ["ticket_id", "answer"],
            "additionalProperties": False,
        },
        "description": "Send an automated answer to the user.",
    },
    "escalate_ticket": {
        "func": escalate_ticket,
        "schema": {
            "type": "object",
            "properties": {"ticket_id": {"type": "string"}},
            "required": ["ticket_id"],
            "additionalProperties": False,
        },
        "description": "Hand the ticket to a human operator.",
    },
}

# ---------- 2. Prompt construction ----------
def build_system_prompt():
    tool_desc = "\n".join(
        f"- `{name}`: {info['description']}"
        for name, info in tools.items()
    )
    return f"""You are an autonomous support agent. You have access to the following tools:
{tool_desc}

When you decide on an action, output JSON with the following schema:
{{
  "type": "call_tool",
  "name": "<tool name>",
  "arguments": <object matching the tool's schema>
}}
If you have reached a final answer, use the tool `reply_ticket`. If you think a human is needed, use `escalate_ticket`. Do not output anything else."""

SYSTEM_PROMPT = build_system_prompt()

# ---------- 3. Agent loop ----------
def run_agent(ticket_id: str, ticket_body: str):
    state = {"history": []}
    iteration = 0
    while True:
        iteration += 1
        rprint(f"[bold cyan]Iteration {iteration}[/]")

        # Build user message
        user_msg = f"""Ticket ID: {ticket_id}
Content:
{ticket_body}

Current state (JSON):
{json.dumps(state, indent=2)}"""

        # Call the model
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_msg},
            ],
            temperature=0,
            response_format={"type": "json_object"},
        )
        action = resp.choices[0].message.parsed
        rprint("[green]Model action:[/]", action)

        # Dispatch
        if action["type"] != "call_tool":
            rprint("[red]Unexpected action type[/]")
            sys.exit(1)

        tool_name = action["name"]
        args = action["arguments"]
        if tool_name not in tools:
            rprint(f"[red]Unknown tool {tool_name}[/]")
            sys.exit(1)

        # Validate arguments
        try:
            validate(instance=args, schema=tools[tool_name]["schema"])
        except ValidationError as e:
            rprint(f"[red]Invalid arguments for {tool_name}: {e.message}[/]")
            sys.exit(1)

        # Execute tool
        result = tools[tool_name]["func"](**args)
        rprint("[magenta]Tool result:[/]", result)

        # Update state
        state["history"].append({"action": action, "result": result})

        # Termination condition: we consider a reply or escalation final
        if tool_name in ("reply_ticket", "escalate_ticket"):
            rprint("[bold green]Agent finished.[/]")
            break

if __name__ == "__main__":
    # Example usage
    ticket_id = "12345"
    ticket_body = """I was charged twice for my last purchase and need a refund."""
    run_agent(ticket_id, ticket_body)

How to run

python agent_loop.py

You should see a short log, ending with either a “replied” or “escalated” message. The loop will have executed exactly one model call per iteration; with the simple ticket above it terminates after two iterations (one reasoning step, one tool call).

When this breaks

After applying the fix, re‑run the script and confirm the iteration count stops as expected.

What it costs

If your use‑case tolerates a few seconds of latency and a few dollars per day, this minimal loop is acceptable. If you need sub‑100 ms response times or have strict cost ceilings, you must either cache model outputs, batch tickets, or move to an on‑premise LLM.

In the interview

Typical interview prompt:

“Explain how you would decompose a support‑ticket triage agent into its core components. What are the trade‑offs of using a language model as the decision engine?”

Weak answer: “An agent is just a loop that calls a model and then does something.” – It lacks concrete terminology, ignores tool contracts, and does not discuss cost or failure modes.

Strong answer:

1. Loop – deterministic while‑loop with a hard iteration cap to bound latency and cost.
2. Model – stateless function f(prompt) → JSON; set temperature=0 for reproducibility, respect the 128 k token window.
3. Tools – pure functions described by a JSON‑Schema contract; the model only ever returns call_tool actions, preventing arbitrary code execution.
4. Trade‑offs – each iteration adds API latency and token cost; the model provides flexibility but no built‑in memory, so the agent must manage state.
5. Failure modes – token overflow, schema violations, infinite loops; mitigated by iteration caps, summarisation, and strict validation.

Senior vs junior expectations:

• Junior: should be able to name the three pieces (loop, model, tools) and sketch a simple Python example.
• Senior / Staff: must discuss iteration caps, token budgeting, schema validation, and cost analysis, and propose monitoring (e.g., Prometheus metrics for iteration count).
• Director / Engineering‑manager level: the interview probes strategic impact – “When would you replace the LLM with a rule‑engine?” – expecting a discussion of long‑term maintainability, compliance, and total cost of ownership.

Follow‑up that separates experience from theory:

“Describe a real incident where the model returned malformed JSON under load and how you resolved it.”

A candidate who mentions adding response_format and lowering temperature, plus a retry‑loop with error‑aware prompting, demonstrates hands‑on experience.

Your tasks

1. Add a summariser tool

   Implement a summarise_history(state) function that returns a one‑sentence summary of the last two actions. Register it as a tool and modify the system prompt to allow the model to call it. Verify that after 5 iterations the prompt size stays under 10 k tokens.

   Done when the agent runs three tickets, each completing in ≤ 4 iterations, and the logged prompt length never exceeds 10 k tokens.

2. Introduce a hard cost budget

   Add a per‑ticket token budget of 500 tokens. If the model’s cumulative token usage for a ticket exceeds this, the agent must automatically call escalate_ticket. Use the usage field from the OpenAI response to track tokens.

   Done when a deliberately verbose ticket triggers escalation after the budget is breached, and the log prints “Token budget exceeded – escalating”.

3. Break and diagnose

   Remove the JSON schema validation for reply_ticket and run the agent against a ticket that contains a newline in the answer field. Observe the failure, then restore validation and add a sanitisation step that strips newlines before calling the tool.

   Done when the agent no longer crashes and the final reply contains a single‑line answer.

4. Scale test

   Write a script that spawns 100 parallel ticket processes (using concurrent.futures.ThreadPoolExecutor). Measure average latency and total cost (use the usage data). Report whether the average latency stays below 500 ms.

   Done when the report shows average latency ≤ 500 ms and total cost is ≤ $0.10 for the batch.

5. Add a retry policy

   Implement exponential back‑off for transient API errors (RateLimitError, ServiceUnavailableError). The agent should retry up to 3 times before giving up and escalating.

   Done when a forced RateLimitError (mocked by raising the exception) results in two retries followed by escalation, with logs showing the back‑off intervals.

Your tasks this week

Do the exercises above before the next chapter. Reading a tutorial and doing
one are different activities and only one of them changes what you can build.

Stuck on any of them? Say so — describe what you tried and what happened:
tell me where you got stuck. I read every one, and the questions
that come back more than twice get answered in the next chapter.

Agent engineering: MCP, subagents and agentic systems

Chapter 1 of 14. New chapter every Sunday morning.
Next: Tool definitions as a contract.

· The full syllabus and every chapter so far
· Subscribers also get the condensed notes for this chapter, the running
recap of everything the series has covered, and the extended guidance:
subscribe

Written by Amit Chakraborty — founding engineer and senior architect: React Native, AI and RAG systems, production architecture. Portfolio · LinkedIn · GitHub.

Need this built, reviewed or taught to your team? Get in touch or email amit@devamit.co.in. Available for senior and founding engineering roles, consulting and training, remote worldwide.