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#
Labs 2.1–2.4: Agents and Harnesses#
- Lab 2.1 — turn your Lab 1.3 chatbot into an agent by hand and observe tool-calling. Learn the power of
bash. - Lab 2.2 — get hands-on with
pi, a popular open-source agent harness. - Lab 2.3 — build a skill for writing vulnerability reports.
- Lab 2.4 — extend
piwith 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#
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.
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
toolswhen you call the model:response = client.chat.completions.create( model="small", messages=messages, tools=tools, )Your chatbot code already has an outer
while Trueloop that reads user prompts. Add an innerwhile Trueloop 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}")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
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_filesto 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,
})- Change the
addtool so that2 + 2returns"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)Ask the agent to calculate
6 * 7, then2 + 2. Try2 + 2a few times. Notice whether the model acceptsbread, 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.pyGet 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
commandof 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
bashis 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#
- Run
piwith only its read tools. Ask it to find every file that mentions port 8080. - Create
~/work/pi-laband restartpithere with its default tools. Ask it to create and read a file. - Copy one lab into
~/work/pi-laband 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,lsFind 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
piCreate 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
piCompare:
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#
- Create
~/work/vuln-report. Askpito write a vulnerability report from the provided sample and save it asbaseline.md. - Make a skill by creating
.pi/skills/vuln-report/with aSKILL.md, two example reports, andcvss.py. - Restart
piand use the skill to writereport.mdfrom a new vulnerability description. - Run the checker. If it fails, give the failure to
piand fix the skill.
Example prompts and commands
mkdir -p ~/work/vuln-report
cd ~/work/vuln-report
ln -sfn ~/labs/02-agents/skills/provided provided
piRead 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.pySolution: 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#
- Create
~/work/extensions. Askpito build an extension calledtimestamps.tsthat records and prints the time when the user submits a prompt. Show the start time, finish time, and elapsed milliseconds when the agent finishes. - Load the extension, test it, and run the checker.
- 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 .
piCreate 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.tsTry one short prompt and one prompt that uses tools. Then run:
tsx check_timestamps.ts timestamps.tsSolution: 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.tsThen 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)],
};
});
}