Building a Minimal Harness
This guide walks through building the simplest possible agentic harness to understand the core concepts. We'll build a CLI agent that can read files, run commands, and have conversations.
Goal
Build a minimal harness with: - Agent loop (think → act → observe) - Tool calling (read_file, write_file, bash) - Basic permission system - Context building - CLI interface
Architecture Overview
minimal-harness/
├── main.py # CLI entry point
├── agent.py # Agent loop
├── tools.py # Tool registry and implementations
├── permissions.py # Permission checking
├── context.py # Context building
├── model.py # Model client
└── config.py # Configuration
Step 1: Model Client
Abstract the model interface first:
# model.py
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
class ModelClient:
def __init__(self, provider="anthropic", model="claude-sonnet-4-20250514"):
self.provider = provider
self.model = model
if provider == "anthropic":
self.client = AsyncAnthropic()
elif provider == "openai":
self.client = AsyncOpenAI()
async def generate(self, messages, tools=None, system=None):
if self.provider == "anthropic":
response = await self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system or "",
messages=messages,
tools=tools or [],
)
return response
# ... handle other providers
Step 2: Tool Registry
# tools.py
import json
import os
import asyncio
from dataclasses import dataclass
@dataclass
class Tool:
name: str
description: str
parameters: dict
risk_level: str # low, medium, high
func: callable
class ToolRegistry:
def __init__(self):
self.tools = {}
def register(self, tool: Tool):
self.tools[tool.name] = tool
def get_schemas(self):
schemas = []
for tool in self.tools.values():
schemas.append({
"name": tool.name,
"description": tool.description,
"input_schema": tool.parameters
})
return schemas
async def execute(self, name, args, context):
tool = self.tools[name]
# Permission check happens here
if not context.permissions.check(tool, args):
return {"error": "Permission denied"}
return await tool.func(args, context)
# Tool implementations
async def read_file(args, ctx):
path = args["path"]
with open(path, 'r') as f:
return {"content": f.read()}
async def write_file(args, ctx):
path = args["path"]
with open(path, 'w') as f:
f.write(args["content"])
return {"success": True, "path": path}
async def bash(args, ctx):
proc = await asyncio.create_subprocess_shell(
args["command"],
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=ctx.working_dir
)
stdout, stderr = await proc.communicate()
return {
"stdout": stdout.decode(),
"stderr": stderr.decode(),
"exitCode": proc.returncode
}
# Register tools
registry = ToolRegistry()
registry.register(Tool("read_file", "Read a file",
{"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]},
"low", read_file))
registry.register(Tool("write_file", "Write a file",
{"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]},
"medium", write_file))
registry.register(Tool("bash", "Run shell command",
{"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]},
"high", bash))
Step 3: Permission System
# permissions.py
import re
class Permissions:
def __init__(self, mode="interactive"):
self.mode = mode
self.allowed = set()
def check(self, tool, args):
if tool.risk_level == "low":
return True
if tool.name in self.allowed:
return True
if self.mode == "yolo":
return True
if self.mode == "interactive":
return self.ask_user(tool, args)
return False
def ask_user(self, tool, args):
print(f"\n🔐 Permission required: {tool.name}")
print(f" Risk: {tool.risk_level}")
print(f" Args: {args}")
response = input(" Allow? [y/N/a(always)]: ").lower()
if response == 'a':
self.allowed.add(tool.name)
return True
return response == 'y'
Step 4: Agent Loop
# agent.py
import json
class Agent:
def __init__(self, model, tools, permissions, working_dir):
self.model = model
self.tools = tools
self.permissions = permissions
self.working_dir = working_dir
self.messages = []
self.system_prompt = """You are a helpful coding assistant.
You have access to tools to read files, write files, and run commands.
Think step by step and use tools when needed."""
async def run(self, user_input):
self.messages.append({"role": "user", "content": user_input})
for turn in range(50): # Max turns
# Build context
context = type('Context', (), {
'permissions': self.permissions,
'working_dir': self.working_dir
})()
# Call model
response = await self.model.generate(
messages=self.messages,
tools=self.tools.get_schemas(),
system=self.system_prompt
)
# Process response
if response.stop_reason == "tool_use":
# Handle tool calls
tool_results = []
for block in response.content:
if block.type == "tool_call":
result = await self.tools.execute(block.name, block.input, context)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
# Add assistant message with tool calls
self.messages.append({"role": "assistant", "content": response.content})
self.messages.append({"role": "user", "content": tool_results})
elif response.stop_reason == "end_turn":
# Extract text response
text = ""
for block in response.content:
if block.type == "text":
text += block.text
self.messages.append({"role": "assistant", "content": response.content})
return text
else:
break
return "Max turns reached"
Step 5: CLI Interface
# main.py
import asyncio
import os
from agent import Agent
from model import ModelClient
from tools import registry
from permissions import Permissions
async def main():
print("Minimal Agentic Harness")
print("Type 'quit' to exit, 'yolo' for auto-approve\n")
model = ModelClient(provider="anthropic")
permissions = Permissions(mode="interactive")
agent = Agent(model, registry, permissions, os.getcwd())
while True:
try:
user_input = input("You: ").strip()
if user_input.lower() == 'quit':
break
if user_input.lower() == 'yolo':
permissions.mode = "yolo"
print("YOLO mode enabled\n")
continue
response = await agent.run(user_input)
print(f"\nAgent: {response}\n")
except KeyboardInterrupt:
break
except Exception as e:
print(f"Error: {e}\n")
if __name__ == "__main__":
asyncio.run(main())
Running It
# Install dependencies
pip install anthropic openai
# Set API key
export ANTHROPIC_API_KEY=your_key
# Run
python main.py
Testing
You: Read the README.md file
Agent: I'll read the README first.
[Permission check for read_file - auto-approved]
[Tool call: read_file with path="README.md"]
[Returns content]
Agent: Here's what the README says...
Extensions to Consider
| Feature | Difficulty | Value |
|---|---|---|
| Streaming responses | Medium | Better UX |
| Context compaction | Medium | Handle long conversations |
| Sub-agents | Hard | Parallelization |
| MCP support | Medium | Extensibility |
| Checkpointing | Medium | Resumability |
| Better permissions | Easy | Safety |
| Diff display | Medium | Better for edits |
| Todo tracking | Easy | Task management |
Key Learnings
- The loop is simple - The complexity comes from all the supporting systems
- Permissions are essential - Don't skip them even in minimal version
- Tool schemas matter - Clear schemas lead to better tool use
- Context management is hard - Easy to bloat quickly
- Error handling is critical - Tools fail, models fail, APIs fail
Next Steps
- Add streaming support
- Implement context compaction
- Add more tools (grep, glob)
- Improve permission system
- Add sub-agent support
- Better error recovery
Related Notes
- Harness Architecture — Full architecture
- Agent Loop — Loop details
- Tool Registry — Tool system
- Permission System — Permissions
- Implementation Guide — More detailed guide
Sources
- Anthropic Cookbook - Tool use examples
- OpenAI Agents SDK quickstart
- LangGraph tutorial
- Building CLI agents patterns