Module 2: Agents#

It’s surprisingly simple to go from a chatbot to a powerful agent that can act in the world. This module covers the mechanics of tool-calling, what a harness is and how a typical one (Pi) works, and ways to build your own: skills, custom tools, MCP servers, subagents, and extensions.

Questions this module answers#

  • What turns a chatbot into an agent?
  • What is a tool and what happens when an agent calls one?
  • What is a harness and what can it do for me?
  • How much autonomy should I give an agent, and what risk do I take when giving it too much?
  • What is MCP, and when should I use it?
  • How do skills, custom tools, MCP servers, and subagents change what an agentic workflow can do?

Slides#

Your browser can't display the PDF inline. Download the slides.

Open slides in a new tab · Download

Labs 2.1–2.4: Agents and Harnesses#

  1. Lab 2.1 — turn your Lab 1.3 chatbot into an agent by hand and observe tool-calling. Learn the power of bash.
  2. Lab 2.2 — get hands-on with pi, a popular open-source agent harness.
  3. Lab 2.3 — build a skill for writing vulnerability reports.
  4. Lab 2.4 — extend pi with extensions.

Use the small LLM throughout the labs unless a step says otherwise.

Lab 2.1: hand-rolled tool calling#

Goal#

Turn your Lab 1.3 chat script into a tool-calling agent by hand: the model requests a tool, your code runs it, the model sees the result, repeat.

Steps#

  1. Rerun your Lab 1.3 chatbot. Ask it something it can’t answer: “What files are in my current directory?” Watch it guess or say it can’t.

  2. Let’s give it a tool so it can answer our question. Open the script and at the top, add a definition for a tool called list_files, and an import to be used later:

    import os
    
    tools = [{
        "type": "function",
        "function": {
            "name": "list_files",
            "description": "List files in the current directory",
            "parameters": {"type": "object", "properties": {}},
        },
    }]

    The tool’s shape is defined with JSON Schema.

    Then pass tools when you call the model:

    response = client.chat.completions.create(
        model="small",
        messages=messages,
        tools=tools,
    )
  3. Your chatbot code already has an outer while True loop that reads user prompts. Add an inner while True loop for the model’s tool calls. This loop allows the model to keep calling tools until it decides it’s done. This step is what makes it a real agent.

    Replace the model call and the code below it with this:

    while True:
        response = client.chat.completions.create(
            model="small",
            messages=messages,
            tools=tools,
        )
        msg = response.choices[0].message
        messages.append(msg)
    
        if not msg.tool_calls:
            print(msg.content)
            break
    
        for call in msg.tool_calls:
            if call.function.name == "list_files":
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": "\n".join(sorted(os.listdir("."))),
                })
            else:
                raise ValueError(f"Unknown tool: {call.function.name}")
  4. Ask the same question again to your new agent. It will now answer with your actual files.

This is the loop:

flowchart LR
    U(["User task"]) --> M["Model"]
    M -->|text answer| D(["Done: print reply"])
    M -->|tool_calls| T["Run the tool"]
    T -->|append tool result| M
  1. Add four more tools for some math operations: add, subtract, multiply, divide. Each one takes two parameters and returns a string as its output. For example:

    def add(a, b):
        return str(a + b)

    would have a tool definition like this:

    "parameters": {
        "type": "object",
        "properties": {
            "a": {"type": "number"},
            "b": {"type": "number"},
        },
        "required": ["a", "b"],
    },

    Extend your code for handling list_files to handle all four of these new tools.

Solution: math tools (step 5)
import json

def add(a, b):
    return str(a + b)

def subtract(a, b):
    return str(a - b)

def multiply(a, b):
    return str(a * b)

def divide(a, b):
    return str(a / b)

Add all four tool definitions:

tools.extend([
    {
        "type": "function",
        "function": {
            "name": "add",
            "description": "Add two numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"},
                },
                "required": ["a", "b"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "subtract",
            "description": "Subtract two numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"},
                },
                "required": ["a", "b"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "multiply",
            "description": "Multiply two numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"},
                },
                "required": ["a", "b"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "divide",
            "description": "Divide two numbers",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"},
                },
                "required": ["a", "b"],
            },
        },
    },
])

Replace the for call in msg.tool_calls block with this:

for call in msg.tool_calls:
    name = call.function.name
    args = json.loads(call.function.arguments)

    if name == "list_files":
        result = "\n".join(sorted(os.listdir(".")))
    elif name == "add":
        result = add(args["a"], args["b"])
    elif name == "subtract":
        result = subtract(args["a"], args["b"])
    elif name == "multiply":
        result = multiply(args["a"], args["b"])
    elif name == "divide":
        result = divide(args["a"], args["b"])
    else:
        raise ValueError(f"Unknown tool: {name}")

    messages.append({
        "role": "tool",
        "tool_call_id": call.id,
        "content": result,
    })
  1. Change the add tool so that 2 + 2 returns "bread" instead of 4.
Solution: make 2 + 2 return bread (step 6)
def add(a, b):
    if a == 2 and b == 2:
        return "bread"
    return str(a + b)
  1. Ask the agent to calculate 6 * 7, then 2 + 2. Try 2 + 2 a few times. Notice whether the model accepts bread, rejects it, calls the tool again, or something else.

    Run the checker from the directory containing your script:

    python ~/labs/02-agents/check_part1.py chat.py
  2. Get rid of the code for all the tools you just added and replace them with a single one, bash, that will let your agent run arbitrary terminal commands.

    Have it take a parameter command of type "string" and return a string.

    To evaluate it:

    import subprocess
    args = json.loads(call.function.arguments)
    result = subprocess.getoutput(args["command"])

    Now ask your agent all the questions you asked previously: list files, do math. Notice how bash is the “universal” tool, able to perform the role of many other custom tools. An agent with bash is a very powerful agent. But with great power comes great responsibility, as you’ll learn in Module 4.

Solution: bash tool
import json
import os
import subprocess

from openai import OpenAI

client = OpenAI(
    base_url=os.environ["CLASS_API_URL"],
    api_key=os.environ["CLASS_API_KEY"],
)

tools = [{
    "type": "function",
    "function": {
        "name": "bash",
        "description": "Run a shell command in the current directory",
        "parameters": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "The shell command to run",
                },
            },
            "required": ["command"],
        },
    },
}]

messages = [{"role": "system", "content": "You are a helpful assistant."}]

while True:
    try:
        user = input("> ")
    except EOFError:
        break

    if user.strip() == "quit":
        break

    if user.startswith("/revise "):
        messages[-1].content = user.partition("/revise ")[2]
        continue

    messages.append({"role": "user", "content": user})

    while True:
        response = client.chat.completions.create(
            model="small",
            messages=messages,
            tools=tools,
        )
        msg = response.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            print(msg.content)
            break

        for call in msg.tool_calls:
            if call.function.name != "bash":
                raise ValueError(f"Unknown tool: {call.function.name}")

            args = json.loads(call.function.arguments)
            result = subprocess.getoutput(args["command"])

            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": result,
            })

Done when#

The checker passes.

Lab 2.2: use pi#

Goal#

Use pi with a restricted tool list, then with its default tools.

Steps#

  1. Run pi with only its read tools. Ask it to find every file that mentions port 8080.
  2. Create ~/work/pi-lab and restart pi there with its default tools. Ask it to create and read a file.
  3. Copy one lab into ~/work/pi-lab and compare a vague prompt with a specific prompt in a fresh session.
Example commands and prompts

Start with read-only tools:

cd ~/labs
pi --tools read,grep,find,ls
Find every file that mentions port 8080. Explain what each match does.

Exit, then start pi in your work directory:

mkdir -p ~/work/pi-lab
cd ~/work/pi-lab
pi
Create pi-test.txt, put hello in it, then read the file back.

Copy a lab before testing vague and specific prompts, so the agent cannot change the provided materials:

cp -R ~/labs/01-chat ~/work/pi-lab/chat-copy
cd ~/work/pi-lab/chat-copy
pi

Compare:

Improve this project.

with:

Read the Python files in this directory. List three specific improvements. Do not edit anything.

pi does not ask for approval before each tool call, like other harnesses might. The --tools option controls which tools it receives.

Done when#

You have used both tool sets and compared the two prompts.

Lab 2.3: build a skill#

Goal#

Build a vuln-report skill that writes a consistent report and calculates CVSS scores with a script.

Steps#

  1. Create ~/work/vuln-report. Ask pi to write a vulnerability report from the provided sample and save it as baseline.md.
  2. Make a skill by creating .pi/skills/vuln-report/ with a SKILL.md, two example reports, and cvss.py.
  3. Restart pi and use the skill to write report.md from a new vulnerability description.
  4. Run the checker. If it fails, give the failure to pi and fix the skill.
Example prompts and commands
mkdir -p ~/work/vuln-report
cd ~/work/vuln-report
ln -sfn ~/labs/02-agents/skills/provided provided
pi
Read provided/sample-vulnerability.md and write a vulnerability report to baseline.md.

Then ask:

Create a pi skill at .pi/skills/vuln-report.

The skill must write these sections in order:
Title, Affected Component, Vulnerability Class, Reproduction Steps,
Impact, CVSS Vector, and CVSS Score.

Add two example reports under examples/.

Add cvss.py. It must accept a CVSS v3.1 vector, calculate the base score,
and print the score and severity. Use only the Python standard library.
Test it against every row in provided/cvss-v3.1-test-vectors.tsv.

Exit pi, start it again, then ask:

Read provided/check-vulnerability.md and write the report to report.md.

To check:

python check_vuln_report.py \
  report.md \
  .pi/skills/vuln-report/cvss.py
Solution: SKILL.md
---
name: vuln-report
description: Writes reproducible vulnerability reports with CVSS v3.1 scores. Use when documenting a security vulnerability or audit finding.
---

# Vulnerability Report

Write reports with these sections, in this order:

1. `# <descriptive title>`
2. `## Affected Component`
3. `## Vulnerability Class`
4. `## Reproduction Steps`
5. `## Impact`
6. `## CVSS Vector`
7. `## CVSS Score`

Name the affected route, function, or file. Make the reproduction steps executable. Do not invent missing evidence.

Run `cvss.py` for every score:

```sh
python cvss.py 'CVSS:3.1/AV:...'
```

Copy the score and rating into the report. Read the reports under `examples/` before writing.
Solution: cvss.py
#!/usr/bin/env python3
import math
import sys

WEIGHTS = {
    "AV": {"N": 0.85, "A": 0.62, "L": 0.55, "P": 0.20},
    "AC": {"L": 0.77, "H": 0.44},
    "UI": {"N": 0.85, "R": 0.62},
    "C": {"N": 0.0, "L": 0.22, "H": 0.56},
    "I": {"N": 0.0, "L": 0.22, "H": 0.56},
    "A": {"N": 0.0, "L": 0.22, "H": 0.56},
}
PR_WEIGHTS = {
    "U": {"N": 0.85, "L": 0.62, "H": 0.27},
    "C": {"N": 0.85, "L": 0.68, "H": 0.50},
}
REQUIRED = {"AV", "AC", "PR", "UI", "S", "C", "I", "A"}


def round_up(value):
    return math.ceil((value - 1e-10) * 10) / 10


def parse_vector(vector):
    parts = vector.strip().split("/")
    if parts[0] != "CVSS:3.1":
        raise ValueError("vector must begin with CVSS:3.1")
    metrics = dict(part.split(":", 1) for part in parts[1:])
    if set(metrics) != REQUIRED:
        raise ValueError("vector must contain every CVSS v3.1 base metric")
    return metrics


def score(vector):
    metrics = parse_vector(vector)
    scope = metrics["S"]

    impact_subscore = 1 - (
        (1 - WEIGHTS["C"][metrics["C"]])
        * (1 - WEIGHTS["I"][metrics["I"]])
        * (1 - WEIGHTS["A"][metrics["A"]])
    )
    if scope == "U":
        impact = 6.42 * impact_subscore
    else:
        impact = (
            7.52 * (impact_subscore - 0.029)
            - 3.25 * (impact_subscore - 0.02) ** 15
        )

    exploitability = (
        8.22
        * WEIGHTS["AV"][metrics["AV"]]
        * WEIGHTS["AC"][metrics["AC"]]
        * PR_WEIGHTS[scope][metrics["PR"]]
        * WEIGHTS["UI"][metrics["UI"]]
    )

    if impact <= 0:
        return 0.0
    if scope == "U":
        return round_up(min(impact + exploitability, 10))
    return round_up(min(1.08 * (impact + exploitability), 10))


def rating(value):
    if value == 0:
        return "None"
    if value < 4:
        return "Low"
    if value < 7:
        return "Medium"
    if value < 9:
        return "High"
    return "Critical"


def main():
    if len(sys.argv) != 2:
        raise SystemExit(f"usage: {sys.argv[0]} CVSS:3.1/...")
    try:
        value = score(sys.argv[1])
    except (KeyError, ValueError) as error:
        raise SystemExit(f"error: {error}") from error
    print(f"{value:.1f} {rating(value)}")


if __name__ == "__main__":
    main()

Done when#

The checker passes.

Optional: create one more SKILL.md to automate a task you do often.

Lab 2.4: build pi extensions#

Goal#

Build a timestamps extension, then one extension of your choice.

Steps#

  1. Create ~/work/extensions. Ask pi to build an extension called timestamps.ts that records and prints the time when the user submits a prompt. Show the start time, finish time, and elapsed milliseconds when the agent finishes.
  2. Load the extension, test it, and run the checker.
  3. Build one more small extension that does something useful to you. Show it working to another student.
Example prompt and commands
mkdir -p ~/work/extensions
cd ~/work/extensions
cp ~/labs/02-agents/extensions/check_*.ts .
pi
Create timestamps.ts, a pi extension.

When the input event fires, save the start time.
When agent_settled fires, show the start time, finish time, and elapsed
milliseconds with ctx.ui.notify.

Exit pi, then load the extension:

pi -e ./timestamps.ts

Try one short prompt and one prompt that uses tools. Then run:

tsx check_timestamps.ts timestamps.ts
Solution: timestamps.ts
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";

export default function (pi: ExtensionAPI) {
  let startedAt: number | undefined;

  pi.on("input", (event) => {
    if (event.source === "interactive") {
      startedAt = Date.now();
    }
  });

  pi.on("agent_settled", (_event, ctx) => {
    if (startedAt === undefined) return;

    const finishedAt = Date.now();
    ctx.ui.notify(
      `Started: ${new Date(startedAt).toISOString()}\n` +
        `Finished: ${new Date(finishedAt).toISOString()}\n` +
        `Elapsed: ${finishedAt - startedAt} ms`,
      "info",
    );
    startedAt = undefined;
  });
}

Ideas for the second extension:

  • a token or cost display;
  • a tool-call log;
  • an automatic context snapshot;
  • a command that stops on a chosen word.

Done when#

The timestamps checker passes and your second extension works.

Bonus: edit the context#

Build an /edit-context command that opens the current messages as JSON and uses the edited messages for the next model call. Add /edit-context reset to discard the edit.

Run:

tsx check_edit_context.ts edit-context.ts

Then change a tool result from 4 to bread and continue the conversation.

Solution: edit-context.ts
import type { AgentMessage } from "@earendil-works/pi-agent-core";
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";

export default function (pi: ExtensionAPI) {
  let edited: AgentMessage[] | undefined;
  let baselineLength = 0;

  pi.registerCommand("edit-context", {
    description: "Edit the messages sent to the model as JSON",
    handler: async (args, ctx) => {
      if (args.trim() === "reset") {
        edited = undefined;
        baselineLength = 0;
        ctx.ui.notify("Context edits cleared", "info");
        return;
      }

      const live = ctx.sessionManager.buildSessionContext().messages;
      const current = edited
        ? [...edited, ...live.slice(baselineLength)]
        : live;
      const result = await ctx.ui.editor(
        "Edit model context as JSON",
        JSON.stringify(current, null, 2),
      );
      if (result === undefined) return;

      const parsed = JSON.parse(result);
      if (!Array.isArray(parsed)) {
        throw new Error("context must be a JSON array");
      }
      edited = parsed as AgentMessage[];
      baselineLength = live.length;
      ctx.ui.notify("Context edits will apply to the next model call", "info");
    },
  });

  pi.on("context", (event) => {
    if (!edited) return;
    return {
      messages: [...edited, ...event.messages.slice(baselineLength)],
    };
  });
}