Module 7: Project — Build a Mini Coding Agent
Add Tools: File and Shell
Description
The agentic loop from the previous capsule works, but the tools are placeholders. In this capsule you replace them with real implementations: read_file, write_file, list_directory, and run_command. You also add basic safety so the agent can't do dangerous things.
Safety First: The WORKSPACE
Before implementing the tools, we define a "sandbox" directory where the agent can operate:
import os
import subprocess
WORKSPACE = os.path.abspath("./test_workspace")
def ensure_workspace():
"""Create the workspace directory if it doesn't exist."""
os.makedirs(WORKSPACE, exist_ok=True)
def safe_path(path: str) -> str:
"""Resolve a path and ensure it's within the workspace."""
resolved = os.path.abspath(os.path.join(WORKSPACE, path))
if not resolved.startswith(WORKSPACE):
raise ValueError(f"Access denied: path '{path}' is outside the workspace")
return resolvedWHAT DOES safe_path DO?
→ Converts relative paths to absolute
→ Verifies that the resulting path is INSIDE the workspace
→ If someone tries "../../etc/passwd" → error
→ It's your FIRST level of security
CONNECTION WITH MODULE 04:
→ This is sandboxing in its simplest form
→ Real coding agents do the same (more sophisticated)
→ The principle of least privilege: access only to the workspace
Implement the 4 Tools
Replace the placeholder execute_tool function with this real version:
def execute_tool(tool_name: str, tool_input: dict) -> str:
"""Execute a tool and return the result as a string."""
try:
if tool_name == "read_file":
return tool_read_file(tool_input["path"])
elif tool_name == "write_file":
return tool_write_file(tool_input["path"], tool_input["content"])
elif tool_name == "list_directory":
return tool_list_directory(tool_input.get("path", "."))
elif tool_name == "run_command":
return tool_run_command(tool_input["command"])
else:
return f"Error: unknown tool '{tool_name}'"
except Exception as e:
return f"Error executing {tool_name}: {str(e)}"Tool 1: read_file
def tool_read_file(path: str) -> str:
"""Read a file from the workspace."""
full_path = safe_path(path)
if not os.path.exists(full_path):
return f"Error: file '{path}' not found"
if not os.path.isfile(full_path):
return f"Error: '{path}' is not a file"
with open(full_path, "r") as f:
content = f.read()
if len(content) > 10000:
content = content[:10000] + f"\n... (truncated, {len(content)} chars total)"
return contentWHAT DOES IT DO?
→ Resolves the path within the workspace (security)
→ Verifies the file exists and is a file
→ Reads the content
→ Truncates if it's very long (protects the context window)
CONNECTION WITH MODULE 02:
→ Truncating to 10,000 chars prevents filling the context window
→ A huge file would consume too many tokens
→ In real agents, this limit is configurable
Tool 2: write_file
def tool_write_file(path: str, content: str) -> str:
"""Write content to a file in the workspace."""
full_path = safe_path(path)
directory = os.path.dirname(full_path)
if directory:
os.makedirs(directory, exist_ok=True)
with open(full_path, "w") as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to '{path}'"WHAT DOES IT DO?
→ Resolves the path within the workspace
→ Creates intermediate directories if they don't exist
→ Writes the content to the file
→ Returns a confirmation with the number of characters
SECURITY NOTE:
→ safe_path prevents writing outside the workspace
→ In a real agent, you'd ask for confirmation before writing
→ Here we simplify it for the educational project
Tool 3: list_directory
def tool_list_directory(path: str = ".") -> str:
"""List contents of a directory in the workspace."""
full_path = safe_path(path)
if not os.path.exists(full_path):
return f"Error: directory '{path}' not found"
if not os.path.isdir(full_path):
return f"Error: '{path}' is not a directory"
entries = []
for entry in sorted(os.listdir(full_path)):
entry_path = os.path.join(full_path, entry)
if os.path.isdir(entry_path):
entries.append(f" {entry}/")
else:
size = os.path.getsize(entry_path)
entries.append(f" {entry} ({size} bytes)")
return f"Contents of '{path}':\n" + "\n".join(entries)WHAT DOES IT DO?
→ Lists files and folders with basic information
→ Distinguishes directories (with /) from files (with size)
→ Sorts alphabetically
CONNECTION WITH MODULE 04:
→ It's the equivalent of list_dir that real agents use
→ It gives the LLM a "map" of the directory
→ The first tool the agent usually uses (exploration)
Tool 4: run_command
ALLOWED_COMMANDS = ["python", "cat", "echo", "ls", "wc", "head", "tail", "grep", "find"]
def tool_run_command(command: str) -> str:
"""Run a shell command with safety restrictions."""
cmd_parts = command.split()
if not cmd_parts:
return "Error: empty command"
base_command = cmd_parts[0]
if base_command not in ALLOWED_COMMANDS:
return (f"Error: command '{base_command}' is not allowed. "
f"Allowed commands: {', '.join(ALLOWED_COMMANDS)}")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30,
cwd=WORKSPACE
)
output = ""
if result.stdout:
output += f"STDOUT:\n{result.stdout}"
if result.stderr:
output += f"STDERR:\n{result.stderr}"
if result.returncode != 0:
output += f"\nReturn code: {result.returncode}"
if not output.strip():
output = "(no output)"
if len(output) > 5000:
output = output[:5000] + f"\n... (truncated)"
return output
except subprocess.TimeoutExpired:
return "Error: command timed out after 30 seconds"WHAT DOES IT DO?
→ Verifies that the command is in the allowed list
→ Runs INSIDE the workspace (cwd=WORKSPACE)
→ Captures stdout and stderr
→ 30-second timeout (prevents infinite loops)
→ Truncates long output
SAFETY:
→ ALLOWED_COMMANDS: only certain commands allowed
→ You can't run rm, sudo, curl, etc.
→ cwd=WORKSPACE: the command runs in the sandbox
→ timeout=30: prevents processes that hang
CONNECTION WITH MODULE 04:
→ This is the permission system in miniature
→ Real agents have similar lists
→ The principle of least privilege applied
Prepare the Test Workspace
Create some files so the agent has something to work with:
mkdir -p test_workspace
cat > test_workspace/hello.py << 'EOF'
def greet(name):
return f"Hello, {name}!"
if __name__ == "__main__":
print(greet("World"))
EOF
cat > test_workspace/calculator.py << 'EOF'
def add(a, b):
return a + b
def subtract(a, b):
return a - b
def multiply(a, b):
return a * b
def divide(a, b):
if b == 0:
return "Error: division by zero"
return a / b
EOF
cat > test_workspace/README.md << 'EOF'
# Test Project
This is a simple test project for the mini coding agent.
## Files
- hello.py: A simple greeting function
- calculator.py: Basic math operations
EOFTest the Complete Agent
Make sure ensure_workspace() is called at the start and run:
python mini_agent.pyTest task 1: Exploration
Describe the task: What files are in the workspace and what does each one do?
Expected result: the agent lists the directory, reads each file, and describes what each one does.
Test task 2: Modification
Describe the task: Add a "power(a, b)" function to calculator.py
that computes a raised to the power of b.
Expected result: it reads calculator.py, adds the function, and possibly verifies by running the file.
Test task 3: Creation
Describe the task: Create a test_calculator.py file with tests
for all the functions in calculator.py.
Expected result: it reads calculator.py to see the functions, creates the test file, and possibly runs them.
What to Observe During the Tests
FOR EACH TASK, DOCUMENT:
1. ITERATIONS
→ How many iterations did it take?
→ Was it efficient or did it go in circles?
2. TOOL SELECTION
→ Which tools did it choose and in what order?
→ Was the sequence logical?
→ Did it ever choose an incorrect tool?
3. REASONING
→ Was the "thinking" (text) useful?
→ Did it explain why it chose each tool?
→ Did it ever reason incorrectly?
4. RESULT
→ Did it complete the task correctly?
→ Were there errors it had to fix?
→ Is the generated code correct?
5. TOKENS
→ How many tokens did it consume in total?
→ Did the history grow quickly?
THESE OBSERVATIONS are the input for capsule 05.
Common Problems and Solutions
PROBLEM: "The LLM requests a tool that doesn't exist"
→ Check the tool definitions — is the description clear?
→ Does the tool's name match exactly?
PROBLEM: "The agent enters an infinite loop"
→ MAX_ITERATIONS should stop it
→ If not, check the stop condition (stop_reason)
PROBLEM: "The agent can't read a file"
→ Is the file INSIDE test_workspace/?
→ Is safe_path working correctly?
PROBLEM: "run_command rejects a command I need"
→ Add the command to ALLOWED_COMMANDS
→ But think about whether it's really safe
PROBLEM: "The output is too long and confusing"
→ Adjust the truncation in log_step (currently 500 chars)
→ Or redirect the output to a log file
The Complete Code: mini_agent.py
Below is the mini_agent.py file assembled with all the pieces from capsules 03 and 04. Copy it complete and replace all the content of your file:
import os
import json
import subprocess
from datetime import datetime
from dotenv import load_dotenv
from anthropic import Anthropic
load_dotenv()
# --- Configuration ---
client = Anthropic()
MODEL = "claude-sonnet-4-20250514"
MAX_ITERATIONS = 20
# Sandbox directory where the agent can operate
WORKSPACE = os.path.abspath("./test_workspace")
SYSTEM_PROMPT = """You are a helpful coding agent. You can read files,
list directories, write files, and run commands to help the user
with coding tasks.
When given a task:
1. First explore to understand the context
2. Plan your approach
3. Execute step by step
4. Verify your work
Always explain what you're doing and why."""
TOOLS = [
{
"name": "read_file",
"description": "Read the contents of a file. Returns the file content as text.",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to the file to read"}
},
"required": ["path"]
}
},
{
"name": "list_directory",
"description": "List files and subdirectories in a directory.",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Directory path. Defaults to '.'", "default": "."}
},
"required": []
}
},
{
"name": "write_file",
"description": "Write content to a file. Creates the file if it doesn't exist, overwrites if it does.",
"input_schema": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Path to write to"},
"content": {"type": "string", "description": "Content to write"}
},
"required": ["path", "content"]
}
},
{
"name": "run_command",
"description": "Run a shell command and return stdout and stderr. Use for running tests, checking files, etc.",
"input_schema": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "The shell command to execute"}
},
"required": ["command"]
}
}
]
# --- Safety ---
def ensure_workspace():
"""Create the workspace directory if it doesn't exist."""
os.makedirs(WORKSPACE, exist_ok=True)
def safe_path(path: str) -> str:
"""Resolve a path and ensure it stays within the workspace (sandboxing)."""
resolved = os.path.abspath(os.path.join(WORKSPACE, path))
if not resolved.startswith(WORKSPACE):
raise ValueError(f"Access denied: path '{path}' is outside the workspace")
return resolved
# --- Tool implementations ---
def tool_read_file(path: str) -> str:
"""Read a file from the workspace."""
full_path = safe_path(path)
if not os.path.exists(full_path):
return f"Error: file '{path}' not found"
if not os.path.isfile(full_path):
return f"Error: '{path}' is not a file"
with open(full_path, "r") as f:
content = f.read()
# Truncate large files to protect the context window
if len(content) > 10000:
content = content[:10000] + f"\n... (truncated, {len(content)} chars total)"
return content
def tool_write_file(path: str, content: str) -> str:
"""Write content to a file in the workspace."""
full_path = safe_path(path)
directory = os.path.dirname(full_path)
if directory:
os.makedirs(directory, exist_ok=True)
with open(full_path, "w") as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to '{path}'"
def tool_list_directory(path: str = ".") -> str:
"""List contents of a directory in the workspace."""
full_path = safe_path(path)
if not os.path.exists(full_path):
return f"Error: directory '{path}' not found"
if not os.path.isdir(full_path):
return f"Error: '{path}' is not a directory"
entries = []
for entry in sorted(os.listdir(full_path)):
entry_path = os.path.join(full_path, entry)
if os.path.isdir(entry_path):
entries.append(f" {entry}/")
else:
size = os.path.getsize(entry_path)
entries.append(f" {entry} ({size} bytes)")
return f"Contents of '{path}':\n" + "\n".join(entries)
ALLOWED_COMMANDS = ["python", "cat", "echo", "ls", "wc", "head", "tail", "grep", "find"]
def tool_run_command(command: str) -> str:
"""Run a shell command with safety restrictions."""
cmd_parts = command.split()
if not cmd_parts:
return "Error: empty command"
base_command = cmd_parts[0]
if base_command not in ALLOWED_COMMANDS:
return (f"Error: command '{base_command}' is not allowed. "
f"Allowed commands: {', '.join(ALLOWED_COMMANDS)}")
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30,
cwd=WORKSPACE
)
output = ""
if result.stdout:
output += f"STDOUT:\n{result.stdout}"
if result.stderr:
output += f"STDERR:\n{result.stderr}"
if result.returncode != 0:
output += f"\nReturn code: {result.returncode}"
if not output.strip():
output = "(no output)"
# Truncate long output to protect the context window
if len(output) > 5000:
output = output[:5000] + f"\n... (truncated)"
return output
except subprocess.TimeoutExpired:
return "Error: command timed out after 30 seconds"
# --- Tool dispatcher ---
def execute_tool(tool_name: str, tool_input: dict) -> str:
"""Route a tool call to the correct implementation."""
try:
if tool_name == "read_file":
return tool_read_file(tool_input["path"])
elif tool_name == "write_file":
return tool_write_file(tool_input["path"], tool_input["content"])
elif tool_name == "list_directory":
return tool_list_directory(tool_input.get("path", "."))
elif tool_name == "run_command":
return tool_run_command(tool_input["command"])
else:
return f"Error: unknown tool '{tool_name}'"
except Exception as e:
return f"Error executing {tool_name}: {str(e)}"
# --- Logging ---
def log_step(iteration: int, step_type: str, content: str):
"""Log each step of the agentic loop for observability."""
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"\n{'='*60}")
print(f"[{timestamp}] Iteration {iteration} | {step_type}")
print(f"{'='*60}")
print(content[:500])
if len(content) > 500:
print(f"... ({len(content)} characters total)")
# --- Agentic loop ---
def run_agent(user_task: str):
"""The core agentic loop: send task → LLM responds → execute tools → repeat."""
print(f"\n{'#'*60}")
print(f"TASK: {user_task}")
print(f"{'#'*60}")
# The history starts with the user's task
messages = [
{"role": "user", "content": user_task}
]
iteration = 0
while iteration < MAX_ITERATIONS:
iteration += 1
log_step(iteration, "LLM REQUEST", f"Sending {len(messages)} messages to the LLM...")
# Step 1: Send to the LLM with the available tools
response = client.messages.create(
model=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
tools=TOOLS,
messages=messages
)
log_step(iteration, "LLM RESPONSE",
f"Stop reason: {response.stop_reason}\n"
f"Tokens - Input: {response.usage.input_tokens}, "
f"Output: {response.usage.output_tokens}")
# Step 2: If stop_reason is "end_turn", the LLM finished
if response.stop_reason == "end_turn":
final_text = ""
for block in response.content:
if hasattr(block, "text"):
final_text += block.text
log_step(iteration, "DONE", final_text)
return final_text
# Step 3: Process the LLM's tool calls
tool_results = []
for block in response.content:
if block.type == "text":
log_step(iteration, "THINKING", block.text)
elif block.type == "tool_use":
log_step(iteration, "TOOL CALL",
f"Tool: {block.name}\nArgs: {json.dumps(block.input, indent=2)}")
# Execute the tool locally
result = execute_tool(block.name, block.input)
log_step(iteration, "TOOL RESULT", result)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
# Step 4: Add the response + results to the history and go back to step 1
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
log_step(iteration, "MAX ITERATIONS", "The iteration limit was reached.")
return "Error: the maximum iteration limit was reached."
if __name__ == "__main__":
ensure_workspace()
task = input("\nDescribe the task for the agent: ")
result = run_agent(task)
print(f"\n{'#'*60}")
print("FINAL RESULT:")
print(result)RUN:
python mini_agent.py
TOTAL: ~250 lines
That's ALL you need for a functional coding agent.
IF YOU USE OPENAI:
Replace the imports, client, and MODEL section.
Adapt run_agent according to the capsule 03 section.
The structure of the loop is identical.
Summary
WHAT YOU IMPLEMENTED:
→ 4 functional tools: read_file, write_file, list_directory, run_command
→ Safety: workspace sandboxing + allowed commands + timeouts + truncation
→ An agent that can explore, read, write, and run code
WHAT YOU LEARNED:
→ Tools are simple Python functions
→ The LLM only REQUESTS, you EXECUTE
→ Safety is the developer's responsibility (you)
→ ~265 lines is enough for a functional agent
CONNECTIONS:
→ Module 02: truncation protects the context window
→ Module 03: real tool calling = JSON from the LLM → Python function
→ Module 04: sandboxing, permissions, file + shell tools
→ Module 05: observing decisions to calibrate confidence
Next capsule: 05 - Analysis and documentation — run the agent on real tasks and document your insights.