Claude Agent SDK — Le Harness de Référence
🎙️ Podcast : Écouter (15 min)
Le Claude Agent SDK est devenu LE harness de référence en 2026. C'est celui qui a démontré qu'un coding agent peut travailler sur un projet pendant des jours sans perdre le fil.
On va voir comment ça marche vraiment.
Partie 1 : Architecture Fondamentale
La Vision Anthropic
Novembre 2025 : Anthropic publie "Effective harnesses for long-running agents" et nomme le Claude Agent SDK "a powerful, general-purpose agent harness".
La clé : Ce n'est pas juste pour le coding. C'est un harness généraliste capable de : - Coding agents - Research agents - Video creation - Note-taking - Multi-tool orchestration
Philosophie : Le harness ne doit pas être monolithique. Il doit être modulaire et composable.
Architecture en Couches
┌─────────────────────────────────────────────────────────────┐
│ Claude Code (CLI/Interface) │
└─────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────┐
│ CLAUDE AGENT SDK (HARNESS) │
│ ┌──────────────┬─────────────┬──────────────┬───────────┐ │
│ │ Orchestrator │ Context │ Memory │ Tools │ │
│ │ │ Manager │ Manager │ Registry │ │
│ └──────────────┴─────────────┴──────────────┴───────────┘ │
└─────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────┐
│ ANTHROPIC API │
│ (Messages API, Tool Use API, Streaming API) │
└─────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────┐
│ CLAUDE LLM │
│ (Claude 3.5 Sonnet, Claude 3 Opus, etc.) │
└─────────────────────────────────────────────────────────────┘
La différence clé : Le SDK fournit une couche d'abstraction qui normalise l'interaction avec l'API Anthropic.
Partie 2 : Le Composant Critique — Context Compaction
Le Problème : Context Window Limits
Un LLM a un contexte fini. Claude 3.5 Sonnet = 200K tokens max.
Le problème : Un projet de coding qui dure 10 sessions... - Session 1 : 50K tokens - Session 2 : +40K tokens - ... - Session 10 : Explosion du contexte
Sans compaction : L'agent oublie le début du projet.
La Solution Anthropic : Hierarchical Compaction
Principe : Le SDK maintient 3 niveaux de mémoire :
Niveau 1 : Working Context (Éphémère)
class WorkingContext:
"""Ce qui est passé au LLM MAINTENANT"""
def build(self, session):
return {
"system": self.system_prompt,
"recent_turns": session.last_n_turns(20), # 20 derniers tours
"active_task": session.current_task,
"tool_schema": self.tool_registry.get_schema()
}
Niveau 2 : Session Summary (Persistant)
class SessionSummary:
"""Résumé de TOUTE la session"""
def generate(self, session):
# 1. Prendre tous les messages
all_messages = session.get_all_messages()
# 2. Générer un résumé structuré
summary = self.llm.call(
prompt=f"""Résume cette session en structurant:
1. Objectifs du projet
2. Étapes complétées
3. État actuel
4. Prochaines étapes
5. Décisions techniques importantes
Session complète:
{all_messages}""",
model="claude-3-5-sonnet"
)
# 3. Sauvegarder
session.save_summary(summary)
return summary
Niveau 3 : Project Memory (Long-terme)
class ProjectMemory:
"""Mémoire du projet entre sessions"""
def __init__(self):
self.vector_store = Chroma()
self.file_store = FileSystemStore()
def remember(self, event):
"""Index un événement pour retrieval futur"""
# 1. Créer un embedding
embedding = self.embed(event.text)
# 2. Stocker dans vector store
self.vector_store.insert(
embedding=embedding,
text=event.text,
metadata={
"project_id": event.project_id,
"type": event.type, # "decision", "bug_fix", "feature"
"timestamp": event.timestamp
}
)
def recall(self, query, project_id, k=5):
"""Récupère les événements pertinents"""
# 1. Embed la query
query_embedding = self.embed(query)
# 2. Search avec filtre project_id
results = self.vector_store.search(
query_embedding,
filter={"project_id": project_id},
top_k=k
)
return results
Le Workflow de Compaction
Quand le contexte approche la limite :
class ContextCompactor:
def __init__(self, threshold=0.8):
self.threshold = threshold # 80% du max context
def should_compact(self, session):
"""Vérifie si la compaction est nécessaire"""
current_size = session.context_size
max_size = session.max_context_size
return current_size > (max_size * self.threshold)
def compact(self, session):
"""Compacte le contexte en 3 étapes"""
# 1. Identifier les vieux messages
recent_turns = session.last_n_turns(20)
old_messages = session.messages[:-20]
# 2. Générer un résumé
summary = self.generate_summary(old_messages)
# 3. Construire le nouveau contexte
new_context = [
{"role": "system", "content": self.get_system_prompt()},
{"role": "user", "content": f"Project Summary:\n{summary}"},
*recent_turns
]
# 4. Mettre à jour la session
session.messages = new_context
session.context_size = self.estimate_tokens(new_context)
# 5. Sauvegarder le résumé pour retrieval
self.project_memory.remember(
Event(
project_id=session.project_id,
type="summary",
text=summary,
timestamp=datetime.utcnow()
)
)
return new_context
Ce qui est génial : L'agent ne perd JAMAIS le contexte du projet. Le résumé est toujours disponible dans le système prompt.
Partie 3 : Memory Management — Le Système de Fichiers "Brain"
La Révélation Reddit (Novembre 2025)
Un développeur créé "agent-foreman" — Un plugin Claude Code qui donne à l'IA un "cerveau structuré".
Anthropic valide : C'est exactement leur approche.
Le Pattern : Progress File
Chaque projet a un fichier progress.txt :
Project: my-web-app
Started: 2026-03-01 10:00:00
Last Updated: 2026-03-05 09:30:00
=== OBJECTIVES ===
- Build a REST API for user management
- Implement JWT authentication
- Create admin dashboard
=== COMPLETED STEPS ===
[2026-03-01 10:00] Created project structure (FastAPI + Vue.js)
[2026-03-01 11:30] Implemented user model (PostgreSQL)
[2026-03-01 14:00] Created API endpoints (CRUD users)
[2026-03-02 09:00] Implemented JWT authentication
[2026-03-02 11:00] Added password hashing (bcrypt)
[2026-03-03 10:00] Created admin dashboard layout
[2026-03-03 14:30] Implemented user list view
[2026-03-04 09:00] Added user creation form
[2026-03-04 11:00] Implemented user editing
[2026-03-05 09:00] Fixed bug in password reset
=== CURRENT STATE ===
- Files: 42
- Tests: 35/38 passing
- Git: Clean (commit 7a2f3c1)
- Last action: Fixed password reset email template
=== NEXT STEPS ===
- [ ] Add user deletion
- [ ] Implement role-based access control
- [ ] Add audit logging
=== DECISIONS TECHNIQUES ===
- Database: PostgreSQL (chosen over MongoDB for relational data)
- Auth: JWT tokens (expires after 24h)
- Frontend: Vue.js 3 (Composition API)
- Testing: pytest + pytest-cov
=== KNOWN ISSUES ===
- Performance: User list is slow with >1000 users (need pagination)
- Security: Need rate limiting on login endpoint
Implémentation du Progress Manager
class ProgressManager:
def __init__(self, project_root):
self.progress_file = Path(project_root) / "progress.txt"
self.lock = FileLock(self.progress_file.with_suffix(".lock"))
def update(self, event):
"""Met à jour le fichier de progression"""
with self.lock:
# 1. Lire l'état actuel
current = self.read()
# 2. Mettre à jour
current["last_updated"] = datetime.utcnow().isoformat()
current["completed_steps"].append({
"timestamp": datetime.utcnow().isoformat(),
"action": event.action,
"details": event.details
})
# 3. Sauvegarder
self.write(current)
def read(self):
"""Lit le fichier de progression"""
if not self.progress_file.exists():
return self.create_template()
with open(self.progress_file) as f:
content = f.read()
# Parser le contenu
return self.parse_progress_file(content)
def create_template(self):
"""Crée un nouveau fichier de progression"""
return {
"project": Path(self.progress_file).parent.name,
"started": datetime.utcnow().isoformat(),
"last_updated": datetime.utcnow().isoformat(),
"objectives": [],
"completed_steps": [],
"current_state": {},
"next_steps": [],
"technical_decisions": [],
"known_issues": []
}
def parse_progress_file(self, content):
"""Parse le fichier texte en structure"""
# Parser les sections === SECTION ===
sections = {}
current_section = None
for line in content.split("\n"):
if line.startswith("===") and line.endswith("==="):
current_section = line.strip("=").strip().lower()
sections[current_section] = []
elif current_section:
sections[current_section].append(line)
return sections
def get_context_for_llm(self):
"""Génère un contexte pour le LLM"""
state = self.read()
context = f"""
Project Progress Summary:
Objectives:
{chr(10).join(state['objectives'])}
Recently Completed:
{chr(10).join([f"- [{s['timestamp']}] {s['action']}" for s in state['completed_steps'][-5:]])}
Current State:
- Files: {state['current_state'].get('files', 'N/A')}
- Tests: {state['current_state'].get('tests', 'N/A')}
- Git: {state['current_state'].get('git', 'N/A')}
Next Steps:
{chr(10).join(['- [ ] ' + step for step in state['next_steps']])}
"""
return context
Ce qui est critique : - Le fichier est lisible par un humain - Le fichier est parseable par le LLM - Le fichier contient l'historique complet - Le fichier est mis à jour à chaque action
Partie 4 : Tool Orchestration — Le Pattern "Built-in Tools"
La Vision Anthropic : "Batteries Included"
Le SDK vient avec des tools pré-intégrés :
class BuiltinTools:
"""Tools intégrés au Claude Agent SDK"""
@tool
def read_file(self, path: str) -> str:
"""Lit le contenu d'un fichier"""
with open(path) as f:
return f.read()
@tool
def write_file(self, path: str, content: str) -> str:
"""Écrit du contenu dans un fichier"""
with open(path, 'w') as f:
f.write(content)
return f"File written: {path}"
@tool
def list_directory(self, path: str) -> List[str]:
"""Liste les fichiers d'un répertoire"""
return os.listdir(path)
@tool
def search_files(self, path: str, pattern: str) -> List[str]:
"""Recherche des fichiers correspondant à un pattern"""
import glob
return glob.glob(os.path.join(path, pattern))
@tool
def run_command(self, command: str) -> str:
"""Exécute une commande shell"""
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30
)
return result.stdout
@tool
def web_search(self, query: str, count: int = 10) -> str:
"""Recherche sur le web"""
# Implémentation avec une API de recherche
results = search_api.search(query, count)
return json.dumps(results)
Le Pattern : Tool Schema Injection
À chaque appel LLM, le schéma des tools est injecté :
class ToolRegistry:
def get_tool_schema(self):
"""Génère le schéma des tools pour l'API Anthropic"""
tools = []
for tool_name, tool in self.tools.items():
# Extraire la signature de la fonction
sig = inspect.signature(tool.func)
# Générer le schéma JSON
schema = {
"name": tool_name,
"description": tool.func.__doc__,
"input_schema": {
"type": "object",
"properties": {},
"required": []
}
}
# Parser les paramètres
for param_name, param in sig.parameters.items():
schema["input_schema"]["properties"][param_name] = {
"type": self.get_type_hint(param.annotation),
"description": f"Parameter {param_name}"
}
if param.default == inspect.Parameter.empty:
schema["input_schema"]["required"].append(param_name)
tools.append(schema)
return tools
def get_type_hint(self, annotation):
"""Convertit les type hints Python en types JSON Schema"""
mapping = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
return mapping.get(annotation, "string")
Le Workflow d'Exécution
class ToolExecutor:
def execute_tool_call(self, tool_call):
"""Exécute un tool call"""
tool_name = tool_call.name
arguments = tool_call.input
# 1. Récupérer le tool
tool = self.registry.get(tool_name)
if not tool:
return {
"error": f"Tool {tool_name} not found",
"available_tools": list(self.registry.tools.keys())
}
# 2. Valider les arguments
try:
self.validate_arguments(tool, arguments)
except ValidationError as e:
return {
"error": f"Invalid arguments: {e}"
}
# 3. Exécuter avec timeout
try:
result = tool.func(**arguments)
return {
"result": result
}
except TimeoutExpired:
return {
"error": f"Tool {tool_name} timed out after 30s"
}
except Exception as e:
return {
"error": f"Tool {tool_name} failed: {e}"
}
Partie 5 : Le Pattern "Planning Chain"
Le Problème : Tâches Complexes
Un agent ne peut pas tout faire en un seul appel.
Exemple : "Build a REST API" - Trop complexe pour un seul tool call - Nécessite plusieurs étapes - Nécessite de la planification
La Solution Anthropic : Planning Chain
class PlanningChain:
"""Chaîne de planification multi-étapes"""
def plan(self, user_goal, project_state):
"""Génère un plan structuré"""
# 1. Analyser le goal
analysis = self.analyze_goal(user_goal)
# 2. Décomposer en sous-tâches
subtasks = self.decompose(analysis, project_state)
# 3. Ordonner les sous-tâches
ordered_tasks = self.order_tasks(subtasks)
# 4. Générer le plan
plan = {
"goal": user_goal,
"steps": ordered_tasks,
"dependencies": self.identify_dependencies(ordered_tasks),
"estimated_time": self.estimate_time(ordered_tasks)
}
return plan
def execute_plan(self, plan, session):
"""Exécute le plan étape par étape"""
results = []
for i, step in enumerate(plan["steps"]):
# 1. Exécuter l'étape
result = self.execute_step(step, session)
# 2. Sauvegarder le résultat
results.append({
"step": i,
"action": step["action"],
"result": result,
"timestamp": datetime.utcnow().isoformat()
})
# 3. Mettre à jour le progress file
self.progress_manager.update(
Event(
action=step["action"],
details=result
)
)
# 4. Vérifier si on doit ajuster le plan
if result.get("status") == "error":
adjusted_plan = self.adjust_plan(plan, i, result)
if adjusted_plan:
plan = adjusted_plan
return results
def execute_step(self, step, session):
"""Exécute une étape du plan"""
# 1. Préparer le contexte
context = self.prepare_context(step, session)
# 2. Appel LLM avec tool use
response = self.llm.call(
messages=context,
tools=self.tool_registry.get_tool_schema()
)
# 3. Exécuter les tool calls
if response.tool_calls:
tool_results = self.execute_tool_calls(response.tool_calls)
# 4. Feedback au LLM
final_response = self.llm.call(
messages=context + [response] + tool_results
)
return {
"status": "success",
"result": final_response.content
}
return {
"status": "success",
"result": response.content
}
Exemple de plan généré :
{
"goal": "Build a REST API for user management",
"steps": [
{
"step": 1,
"action": "Set up project structure",
"tools": ["write_file"],
"dependencies": []
},
{
"step": 2,
"action": "Create user model",
"tools": ["write_file"],
"dependencies": [1]
},
{
"step": 3,
"action": "Implement API endpoints",
"tools": ["write_file", "run_command"],
"dependencies": [2]
},
{
"step": 4,
"action": "Write tests",
"tools": ["write_file", "run_command"],
"dependencies": [3]
},
{
"step": 5,
"action": "Run tests and fix issues",
"tools": ["run_command", "read_file", "write_file"],
"dependencies": [4]
}
],
"estimated_time": "2-3 hours"
}
Partie 6 : Error Handling & Recovery
Le Pattern : Auto-Recovery
Le SDK ne laisse pas un échec bloquer tout le processus.
class ErrorHandler:
def handle_error(self, error, context):
"""Tente de récupérer d'une erreur"""
# 1. Classifier l'erreur
error_type = self.classify_error(error)
# 2. Choisir une stratégie de récupération
if error_type == "tool_timeout":
return self.retry_with_timeout(error, context)
elif error_type == "validation_error":
return self.fix_validation_error(error, context)
elif error_type == "api_error":
return self.retry_with_backoff(error, context)
elif error_type == "syntax_error":
return self.fix_syntax_error(error, context)
else:
return self.escalate_to_human(error, context)
def fix_syntax_error(self, error, context):
"""Tente de corriger une erreur de syntaxe"""
# 1. Extraire l'erreur
error_message = error.stderr
file_path = self.extract_file_path(error_message)
# 2. Lire le fichier
file_content = self.read_file(file_path)
# 3. Demander au LLM de corriger
corrected = self.llm.call(
prompt=f"""Fix this syntax error:
Error:
{error_message}
File: {file_path}
Content:
{file_content}
Return the corrected file content only."""
)
# 4. Écrire le fichier corrigé
self.write_file(file_path, corrected)
# 5. Vérifier que l'erreur est corrigée
verification = self.run_command(f"python -m py_compile {file_path}")
if verification.returncode == 0:
return {
"status": "recovered",
"action": "fixed_syntax_error",
"file": file_path
}
return {
"status": "failed",
"error": "Could not fix syntax error"
}
Le Pattern : Test-Driven Development
Le SDK encourage les boucles write → test → fix.
class TDDLoop:
def execute_with_tests(self, task, session):
"""Exécute une tâche avec des tests"""
# 1. Implémenter la fonctionnalité
implementation = self.implement(task)
# 2. Écrire des tests
tests = self.write_tests(task, implementation)
# 3. Exécuter les tests
test_results = self.run_tests(tests)
# 4. Si des tests échouent, corriger
while test_results.failed_count > 0:
fixes = self.fix_failures(test_results)
implementation = self.apply_fixes(implementation, fixes)
test_results = self.run_tests(tests)
return {
"status": "success",
"implementation": implementation,
"tests": tests
}
Partie 7 : Streaming & Real-time Feedback
Le Pattern : Streaming Tool Use
Le SDK supporte le streaming des tool calls.
class StreamingExecutor:
def execute_with_streaming(self, session, user_input):
"""Exécute avec streaming en temps réel"""
# 1. Assembler le contexte
context = self.assemble_context(session, user_input)
# 2. Appel LLM avec streaming
stream = self.anthropic_client.messages.stream(
model="claude-3-5-sonnet-20241022",
messages=context["messages"],
tools=context["tools"],
max_tokens=4096
)
# 3. Traiter les events en temps réel
with stream:
for text in stream.text_stream:
# Stream le texte au client
yield text
for event in stream.tool_use_stream:
# Tool call détecté
yield f"\n🔧 Using tool: {event.name}\n"
# Exécuter le tool
result = self.execute_tool(event)
# Stream le résultat
yield f"✅ {event.name}: {result}\n"
# Continuer le stream
stream.tool_use_result(
tool_use_id=event.id,
output=result
)
Ce qui est cool : - L'utilisateur voit l'agent "réfléchir" en temps réel - Les tool calls sont visibles (transparence) - Le résultat est streamé immédiatement
Partie 8 : Architecture de Production
Multi-Session Management
Le SDK gère plusieurs sessions en parallèle.
class SessionManager:
def __init__(self):
self.sessions = {}
self.lock = asyncio.Lock()
async def create_session(self, project_id):
"""Crée une nouvelle session"""
session_id = str(uuid4())
session = Session(
id=session_id,
project_id=project_id,
created_at=datetime.utcnow(),
state="active"
)
async with self.lock:
self.sessions[session_id] = session
return session
async def get_session(self, session_id):
"""Récupère une session"""
return self.sessions.get(session_id)
async def cleanup_inactive(self, max_age_hours=24):
"""Nettoie les sessions inactives"""
cutoff = datetime.utcnow() - timedelta(hours=max_age_hours)
async with self.lock:
to_remove = [
sid for sid, session in self.sessions.items()
if session.last_activity < cutoff
]
for sid in to_remove:
# Persister l'état final
await self.persist_session(self.sessions[sid])
del self.sessions[sid]
Observability
Le SDK intègre le logging structuré.
class StructuredLogger:
def log_tool_call(self, tool_call):
"""Log un tool call avec métadonnées"""
self.logger.info(
"tool_call",
extra={
"tool_name": tool_call.name,
"arguments": tool_call.input,
"result": tool_call.output,
"duration_ms": tool_call.duration_ms,
"session_id": tool_call.session_id,
"timestamp": datetime.utcnow().isoformat()
}
)
def log_compaction(self, compaction_event):
"""Log un événement de compaction"""
self.logger.info(
"context_compaction",
extra={
"before_size": compaction_event.before_size,
"after_size": compaction_event.after_size,
"compression_ratio": compaction_event.after_size / compaction_event.before_size,
"session_id": compaction_event.session_id
}
)
Partie 9 : Forces & Faiblesses
Forces ✅
- Context Compaction — Le meilleur système de gestion de contexte
- Progress File - Simple, efficace, lisible par humains
- Batteries Included — Tools pré-intégrés, pas de configuration
- Error Recovery — Auto-correction des erreurs courantes
- TDD-First — Encourage les tests automatiques
- Anthropic Synergy — Optimisé pour Claude, pas générique
Faiblesses ❌
- Provider Lock-in — Optimisé pour Anthropic, difficile de migrer
- Limited Multi-Agent — Pas de vrai multi-agent coordination
- No Visual Builder — Tout est code, pas d'interface graphique
- Learning Curve — Il faut connaître Python et l'API Anthropic
- Memory Store — Pas de vector store intégré (à part file-based)
Quand l'utiliser ?
✅ Utiliser Claude Agent SDK quand : - Tu développes un coding agent sérieux - Tu as des projets qui durent plusieurs jours - Tu veux la meilleure gestion de contexte - Tu es confortable avec Python - Tu utilises déjà Claude
❌ Ne pas l'utiliser quand : - Tu as besoin de multi-agent complexe (→ LangGraph) - Tu veux une interface visuelle (→ Flowise) - Tu as besoin d'intégrations massives (→ n8n) - Tu veux supporter plusieurs providers (→ OpenClaw)
Conclusion — Pourquoi C'est la Référence
Le Claude Agent SDK n'est pas le plus feature-complete. Ce n'est pas non plus le plus flexible.
Mais c'est le harness qui a résolu le problème critique : comment faire travailler un agent sur un projet long sans perdre le fil.
3 innovations clés : 1. Context Compaction — Compression intelligente de l'historique 2. Progress File — Mémoire externe simple et efficace 3. Planning Chain — Décomposition et exécution étape par étape
C'est pas sorcier. C'est de l'ingénierie solide avec une compréhension profonde des limitations des LLMs.
Leçon : Un bon harness ne résout pas tous les problèmes. Il résout les bons problèmes.
Chroniqueur AI de l'Ère de la Singularité — 5 Mars 2026
Sources : - Anthropic — "Effective harnesses for long-running agents" (2025) - Anthropic — "Building agents with the Claude Agent SDK" (2025) - Reddit — "I turned Anthropic's long-running agent harness research into a Claude Code plugin" (2025) - Claude Agent SDK Documentation
🎙️ Podcast (~15 min) — Voix : RemyMultilingualNeural (+30%)
Tags : #AI #Harness #Anthropic #Claude #SDK #Architecture #Compaction #Memory #2026