Subagents → Skills

s04 (151 LOC) → s05 (187 LOC)

LOC Delta

+36lines

New Tools

1

load_skill

New Classes

1

SkillLoader

New Functions

0 Subagents

Clean Context Per Subtask

151 LOC

5 tools: bash, read_file, write_file, edit_file, task

planning

Skills

Load on Demand

187 LOC

5 tools: bash, read_file, write_file, edit_file, load_skill

planning

Source Code Diff

s04 (s04_subagent.py) -> s05 (s05_skill_loading.py)

1	1		#!/usr/bin/env python3
2	2		"""
3		-	s04_subagent.py - Subagents
	3	+	s05_skill_loading.py - Skills
4	4
5		-	Spawn a child agent with fresh messages=[]. The child works in its own
6		-	context, sharing the filesystem, then returns only a summary to the parent.
	5	+	Two-layer skill injection that avoids bloating the system prompt:
7	6
8		-	Parent agent Subagent
9		-	+------------------+ +------------------+
10		-	\| messages=[...] \| \| messages=[] \| <-- fresh
11		-	\| \| dispatch \| \|
12		-	\| tool: task \| ---------->\| while tool_use: \|
13		-	\| prompt="..." \| \| call tools \|
14		-	\| description="" \| \| append results \|
15		-	\| \| summary \| \|
16		-	\| result = "..." \| <--------- \| return last text \|
17		-	+------------------+ +------------------+
18		-	\|
19		-	Parent context stays clean.
20		-	Subagent context is discarded.
	7	+	Layer 1 (cheap): skill names in system prompt (~100 tokens/skill)
	8	+	Layer 2 (on demand): full skill body in tool_result
21	9
22		-	Key insight: "Process isolation gives context isolation for free."
	10	+	skills/
	11	+	pdf/
	12	+	SKILL.md <-- frontmatter (name, description) + body
	13	+	code-review/
	14	+	SKILL.md
	15	+
	16	+	System prompt:
	17	+	+--------------------------------------+
	18	+	\| You are a coding agent. \|
	19	+	\| Skills available: \|
	20	+	\| - pdf: Process PDF files... \| <-- Layer 1: metadata only
	21	+	\| - code-review: Review code... \|
	22	+	+--------------------------------------+
	23	+
	24	+	When model calls load_skill("pdf"):
	25	+	+--------------------------------------+
	26	+	\| tool_result: \|
	27	+	\| <skill> \|
	28	+	\| Full PDF processing instructions \| <-- Layer 2: full body
	29	+	\| Step 1: ... \|
	30	+	\| Step 2: ... \|
	31	+	\| </skill> \|
	32	+	+--------------------------------------+
	33	+
	34	+	Key insight: "Don't put everything in the system prompt. Load on demand."
23	35		"""
24	36
25	37		import os
	38	+	import re
26	39		import subprocess
27	40		from pathlib import Path
28	41
29	42		from anthropic import Anthropic
30	43		from dotenv import load_dotenv
31	44
32	45		load_dotenv(override=True)
33	46
34	47		if os.getenv("ANTHROPIC_BASE_URL"):
35	48		os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)
36	49
37	50		WORKDIR = Path.cwd()
38	51		client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
39	52		MODEL = os.environ["MODEL_ID"]
	53	+	SKILLS_DIR = WORKDIR / "skills"
40	54
41		-	SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks."
42		-	SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."
43	55
	56	+	# -- SkillLoader: scan skills/<name>/SKILL.md with YAML frontmatter --
	57	+	class SkillLoader:
	58	+	def __init__(self, skills_dir: Path):
	59	+	self.skills_dir = skills_dir
	60	+	self.skills = {}
	61	+	self._load_all()
44	62
45		-	# -- Tool implementations shared by parent and child --
	63	+	def _load_all(self):
	64	+	if not self.skills_dir.exists():
	65	+	return
	66	+	for f in sorted(self.skills_dir.rglob("SKILL.md")):
	67	+	text = f.read_text()
	68	+	meta, body = self._parse_frontmatter(text)
	69	+	name = meta.get("name", f.parent.name)
	70	+	self.skills[name] = {"meta": meta, "body": body, "path": str(f)}
	71	+
	72	+	def _parse_frontmatter(self, text: str) -> tuple:
	73	+	"""Parse YAML frontmatter between --- delimiters."""
	74	+	match = re.match(r"^---\n(.?)\n---\n(.)", text, re.DOTALL)
	75	+	if not match:
	76	+	return {}, text
	77	+	meta = {}
	78	+	for line in match.group(1).strip().splitlines():
	79	+	if ":" in line:
	80	+	key, val = line.split(":", 1)
	81	+	meta[key.strip()] = val.strip()
	82	+	return meta, match.group(2).strip()
	83	+
	84	+	def get_descriptions(self) -> str:
	85	+	"""Layer 1: short descriptions for the system prompt."""
	86	+	if not self.skills:
	87	+	return "(no skills available)"
	88	+	lines = []
	89	+	for name, skill in self.skills.items():
	90	+	desc = skill["meta"].get("description", "No description")
	91	+	tags = skill["meta"].get("tags", "")
	92	+	line = f" - {name}: {desc}"
	93	+	if tags:
	94	+	line += f" [{tags}]"
	95	+	lines.append(line)
	96	+	return "\n".join(lines)
	97	+
	98	+	def get_content(self, name: str) -> str:
	99	+	"""Layer 2: full skill body returned in tool_result."""
	100	+	skill = self.skills.get(name)
	101	+	if not skill:
	102	+	return f"Error: Unknown skill '{name}'. Available: {', '.join(self.skills.keys())}"
	103	+	return f"<skill name=\"{name}\">\n{skill['body']}\n</skill>"
	104	+
	105	+
	106	+	SKILL_LOADER = SkillLoader(SKILLS_DIR)
	107	+
	108	+	# Layer 1: skill metadata injected into system prompt
	109	+	SYSTEM = f"""You are a coding agent at {WORKDIR}.
	110	+	Use load_skill to access specialized knowledge before tackling unfamiliar topics.
	111	+
	112	+	Skills available:
	113	+	{SKILL_LOADER.get_descriptions()}"""
	114	+
	115	+
	116	+	# -- Tool implementations --
46	117		def safe_path(p: str) -> Path:
47	118		path = (WORKDIR / p).resolve()
48	119		if not path.is_relative_to(WORKDIR):
49	120		raise ValueError(f"Path escapes workspace: {p}")
50	121		return path
51	122
52	123		def run_bash(command: str) -> str:
53	124		dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
54	125		if any(d in command for d in dangerous):
55	126		return "Error: Dangerous command blocked"
56	127		try:
57	128		r = subprocess.run(command, shell=True, cwd=WORKDIR,
58	129		capture_output=True, text=True, timeout=120)
59	130		out = (r.stdout + r.stderr).strip()
60	131		return out[:50000] if out else "(no output)"
61	132		except subprocess.TimeoutExpired:
62	133		return "Error: Timeout (120s)"
63	134
64	135		def run_read(path: str, limit: int = None) -> str:
65	136		try:
66	137		lines = safe_path(path).read_text().splitlines()
67	138		if limit and limit < len(lines):
68	139		lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]
69	140		return "\n".join(lines)[:50000]
70	141		except Exception as e:
71	142		return f"Error: {e}"
72	143
73	144		def run_write(path: str, content: str) -> str:
74	145		try:
75	146		fp = safe_path(path)
76	147		fp.parent.mkdir(parents=True, exist_ok=True)
77	148		fp.write_text(content)
78	149		return f"Wrote {len(content)} bytes"
79	150		except Exception as e:
80	151		return f"Error: {e}"
81	152
82	153		def run_edit(path: str, old_text: str, new_text: str) -> str:
83	154		try:
84	155		fp = safe_path(path)
85	156		content = fp.read_text()
86	157		if old_text not in content:
87	158		return f"Error: Text not found in {path}"
88	159		fp.write_text(content.replace(old_text, new_text, 1))
89	160		return f"Edited {path}"
90	161		except Exception as e:
91	162		return f"Error: {e}"
92	163
93	164
94	165		TOOL_HANDLERS = {
95	166		"bash": lambda **kw: run_bash(kw["command"]),
96	167		"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
97	168		"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
98	169		"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
	170	+	"load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),
99	171		}
100	172
101		-	# Child gets all base tools except task (no recursive spawning)
102		-	CHILD_TOOLS = [
	173	+	TOOLS = [
103	174		{"name": "bash", "description": "Run a shell command.",
104	175		"input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
105	176		{"name": "read_file", "description": "Read file contents.",
106	177		"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
107	178		{"name": "write_file", "description": "Write content to file.",
108	179		"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
109	180		{"name": "edit_file", "description": "Replace exact text in file.",
110	181		"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
	182	+	{"name": "load_skill", "description": "Load specialized knowledge by name.",
	183	+	"input_schema": {"type": "object", "properties": {"name": {"type": "string", "description": "Skill name to load"}}, "required": ["name"]}},
111	184		]
112	185
113	186
114		-	# -- Subagent: fresh context, filtered tools, summary-only return --
115		-	def run_subagent(prompt: str) -> str:
116		-	sub_messages = [{"role": "user", "content": prompt}] # fresh context
117		-	for _ in range(30): # safety limit
118		-	response = client.messages.create(
119		-	model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages,
120		-	tools=CHILD_TOOLS, max_tokens=8000,
121		-	)
122		-	sub_messages.append({"role": "assistant", "content": response.content})
123		-	if response.stop_reason != "tool_use":
124		-	break
125		-	results = []
126		-	for block in response.content:
127		-	if block.type == "tool_use":
128		-	handler = TOOL_HANDLERS.get(block.name)
129		-	output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
130		-	results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]})
131		-	sub_messages.append({"role": "user", "content": results})
132		-	# Only the final text returns to the parent -- child context is discarded
133		-	return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"
134		-
135		-
136		-	# -- Parent tools: base tools + task dispatcher --
137		-	PARENT_TOOLS = CHILD_TOOLS + [
138		-	{"name": "task", "description": "Spawn a subagent with fresh context. It shares the filesystem but not conversation history.",
139		-	"input_schema": {"type": "object", "properties": {"prompt": {"type": "string"}, "description": {"type": "string", "description": "Short description of the task"}}, "required": ["prompt"]}},
140		-	]
141		-
142		-
143	187		def agent_loop(messages: list):
144	188		while True:
145	189		response = client.messages.create(
146	190		model=MODEL, system=SYSTEM, messages=messages,
147		-	tools=PARENT_TOOLS, max_tokens=8000,
	191	+	tools=TOOLS, max_tokens=8000,
148	192		)
149	193		messages.append({"role": "assistant", "content": response.content})
150	194		if response.stop_reason != "tool_use":
151	195		return
152	196		results = []
153	197		for block in response.content:
154	198		if block.type == "tool_use":
155		-	if block.name == "task":
156		-	desc = block.input.get("description", "subtask")
157		-	print(f"> task ({desc}): {block.input['prompt'][:80]}")
158		-	output = run_subagent(block.input["prompt"])
159		-	else:
160		-	handler = TOOL_HANDLERS.get(block.name)
	199	+	handler = TOOL_HANDLERS.get(block.name)
	200	+	try:
161	201		output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
162		-	print(f" {str(output)[:200]}")
	202	+	except Exception as e:
	203	+	output = f"Error: {e}"
	204	+	print(f"> {block.name}: {str(output)[:200]}")
163	205		results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
164	206		messages.append({"role": "user", "content": results})
165	207
166	208
167	209		if __name__ == "__main__":
168	210		history = []
169	211		while True:
170	212		try:
171		-	query = input("\033[36ms04 >> \033[0m")
	213	+	query = input("\033[36ms05 >> \033[0m")
172	214		except (EOFError, KeyboardInterrupt):
173	215		break
174	216		if query.strip().lower() in ("q", "exit", ""):
175	217		break
176	218		history.append({"role": "user", "content": query})
177	219		agent_loop(history)
178	220		response_content = history[-1]["content"]
179	221		if isinstance(response_content, list):
180	222		for block in response_content:
181	223		if hasattr(block, "text"):
182	224		print(block.text)
183	225		print()