Chapter 24
Files API: Cross-Request PDF/Image Reuse, 500MB File Handling and ZDR Limitations
Chapter 24: Advisor Tool: Claude's Metacognition and Task Planning Capabilities
24.1 The Value of Metacognition in AI Agents
Metacognition — "cognition about cognition" — is the ability to reflect on one's own thinking process, evaluate one's capability boundaries, and plan execution strategies. For an AI agent, metacognitive capability means being able to answer the following questions before executing a task:
- Have I fully understood this task?
- What steps are needed to complete it?
- Do I have the tools needed for each step?
- Which steps carry the highest risk and require extra caution?
- If a step fails, how do I recover?
The Advisor Tool is a design pattern that makes Claude's metacognitive capability explicit as observable, auditable tool calls, rendering the task planning process transparent and controllable.
24.2 The Design Philosophy of the Advisor Tool
The core idea of the Advisor Tool is: before Claude starts acting, have it first review the entire task as an "advisor".
This pattern draws inspiration from the human "dual-system" decision theory:
- System 1 (fast, intuitive): Start executing immediately, relying on experience for quick responses
- System 2 (slow, analytical): Stop first, analyze systematically, then act
For complex agent tasks, jumping directly into "System 1" execution mode often leads to:
- Discovering missing critical information mid-task
- Wrong tool call ordering
- Ignoring edge cases and corrupting data
- Irreversible operations (e.g., accidentally sending test emails to real users)
The Advisor Tool forces Claude into "System 2" mode first.
24.3 Complete Advisor Tool Definition
advisor_tool = {
"name": "advisor",
"description": """Call this tool to perform metacognitive assessment and task planning
before executing any complex task.
You SHOULD use this tool when:
1. The task involves multiple steps or combinations of multiple tools
2. The task involves irreversible operations (delete, send, submit, etc.)
3. Task requirements are unclear and need clarification
4. The cost of task failure is high
This tool performs NO actual operations — it only plans and evaluates.
[Advisor evaluation output includes]
- task_understanding: Understanding of the task and potential ambiguities
- execution_plan: Detailed steps with ordering and dependencies
- risk_assessment: Risk level and mitigations for each step
- capability_check: List of tools and information needed
- clarifications_needed: Questions to ask the user (if any)
- recommendation: proceed / clarify / reject""",
"input_schema": {
"type": "object",
"properties": {
"task_description": {
"type": "string",
"description": "Describe the task in your own words (not a repeat of the user's message)"
},
"task_complexity": {
"type": "string",
"enum": ["simple", "moderate", "complex", "critical"],
"description": "Complexity: simple (1 step) / moderate (multi-step) / complex (multi-tool multi-step) / critical (high-risk or irreversible)"
},
"proposed_steps": {
"type": "array",
"description": "List of planned execution steps",
"items": {
"type": "object",
"properties": {
"step_number": {"type": "integer"},
"action": {"type": "string"},
"tool_required": {"type": "string"},
"dependencies": {
"type": "array",
"items": {"type": "integer"},
"description": "Step numbers this step depends on"
},
"risk_level": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
},
"reversible": {
"type": "boolean",
"description": "Whether this step can be undone"
}
},
"required": ["step_number", "action", "risk_level", "reversible"]
}
},
"missing_information": {
"type": "array",
"items": {"type": "string"},
"description": "List of critical information missing to execute the task (if any)"
},
"potential_issues": {
"type": "array",
"items": {"type": "string"},
"description": "Anticipated potential problems or risks"
},
"recommendation": {
"type": "string",
"enum": ["proceed", "clarify", "reject"],
"description": "proceed: sufficient info, ready to execute / clarify: need user input / reject: refuse execution (see rejection_reason)"
},
"rejection_reason": {
"type": "string",
"description": "If recommendation=reject, state the reason"
}
},
"required": ["task_description", "task_complexity", "proposed_steps", "recommendation"]
}
}
24.4 Advisor-Executor Architecture
The Advisor Tool is typically used in combination with execution tools, forming a two-phase "plan first, execute second" architecture:
import anthropic
import json
from typing import List, Dict, Any
class AdvisorExecutorAgent:
"""Agent implementing the Advisor-Executor dual-phase architecture"""
def __init__(
self,
execution_tools: List[dict],
tool_functions: Dict[str, callable],
model: str = "claude-opus-4-5",
require_approval_for: List[str] = None
):
self.client = anthropic.Anthropic()
self.model = model
self.execution_tools = execution_tools
self.tool_functions = tool_functions
self.require_approval_for = require_approval_for or ["high", "critical"]
self.advisor_tools = [advisor_tool]
self.all_tools = [advisor_tool] + execution_tools
self.advisor_result = None
def _run_advisor_phase(self, user_message: str, system: str) -> dict:
"""Run the Advisor phase to obtain an execution plan"""
print("\n" + "="*50)
print("ADVISOR PHASE: Planning task...")
print("="*50)
messages = [{"role": "user", "content": user_message}]
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system + "\n\nBefore taking any action, you MUST call the advisor tool to plan the task.",
tools=self.advisor_tools,
tool_choice={"type": "tool", "name": "advisor"},
messages=messages
)
for block in response.content:
if block.type == "tool_use" and block.name == "advisor":
self.advisor_result = block.input
print(f"\nTask complexity: {block.input.get('task_complexity', 'unknown')}")
print(f"Recommendation: {block.input.get('recommendation', 'unknown')}")
steps = block.input.get('proposed_steps', [])
print(f"\nExecution plan ({len(steps)} steps):")
risk_symbols = {"low": "[OK]", "medium": "[~]", "high": "[!]", "critical": "[X]"}
for step in steps:
sym = risk_symbols.get(step.get("risk_level", "low"), "[?]")
print(f" Step {step['step_number']} {sym} {step['action']}")
issues = block.input.get('potential_issues', [])
if issues:
print(f"\nPotential issues:")
for issue in issues:
print(f" - {issue}")
missing = block.input.get('missing_information', [])
if missing:
print(f"\nMissing information:")
for info in missing:
print(f" - {info}")
return block.input
return {}
def _needs_approval(self, advisor_result: dict) -> bool:
steps = advisor_result.get('proposed_steps', [])
return any(s.get('risk_level') in self.require_approval_for for s in steps)
def _request_approval(self, advisor_result: dict) -> bool:
print("\n" + "! "*20)
print("HIGH-RISK STEPS DETECTED: Human approval required")
print("! "*20)
risky_steps = [
s for s in advisor_result.get('proposed_steps', [])
if s.get('risk_level') in self.require_approval_for
]
print("\nHigh-risk steps:")
for step in risky_steps:
print(f" - Step {step['step_number']}: {step['action']}")
print(f" Risk: {step['risk_level']}, Reversible: {step.get('reversible', 'unknown')}")
response = input("\nApprove execution? (yes/no): ").strip().lower()
return response in ['yes', 'y']
def _run_execution_phase(self, user_message: str, system: str,
advisor_result: dict) -> str:
"""Run the execution phase"""
print("\n" + "="*50)
print("EXECUTION PHASE: Running task...")
print("="*50)
plan_summary = json.dumps(advisor_result, indent=2)
enhanced_system = f"""{system}
You have created the following execution plan via the advisor tool:
{plan_summary}
Execute strictly according to the plan. If you encounter unexpected situations during execution, stop and re-evaluate."""
messages = [{"role": "user", "content": user_message}]
while True:
response = self.client.messages.create(
model=self.model,
max_tokens=8192,
system=enhanced_system,
tools=self.all_tools,
messages=messages
)
if response.stop_reason == "end_turn":
return ' '.join(b.text for b in response.content if b.type == "text")
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
if block.name == "advisor":
print(f"\n[Re-planning] Reassessing due to unexpected situation...")
result_content = json.dumps({
"acknowledged": True,
"message": "New plan recorded; please continue execution"
})
else:
print(f"\n[Execute] {block.name}: {json.dumps(block.input)[:100]}")
try:
result = self.tool_functions[block.name](**block.input)
result_content = json.dumps(result, default=str)
print(f"[Success] {block.name}")
except Exception as e:
result_content = json.dumps({
"error": str(e), "tool": block.name
})
print(f"[Failed] {block.name}: {e}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result_content
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
break
return "Execution complete"
def run(self, user_message: str, system: str = "") -> str:
"""Full Advisor-Executor workflow"""
default_system = """You are a careful and systematic AI assistant.
Before executing any complex task, you plan systematically, assess risks,
and ensure nothing goes wrong."""
effective_system = system or default_system
# Phase 1: Planning
advisor_result = self._run_advisor_phase(user_message, effective_system)
if not advisor_result:
return "Planning phase failed; cannot proceed"
recommendation = advisor_result.get('recommendation', 'proceed')
if recommendation == 'reject':
reason = advisor_result.get('rejection_reason', 'Task is not suitable for automated execution')
return f"Task rejected. Reason: {reason}"
if recommendation == 'clarify':
missing = advisor_result.get('missing_information', [])
questions = '\n'.join(f" - {q}" for q in missing)
return f"The following information is needed before proceeding:\n{questions}"
# Check for human approval requirement
if self._needs_approval(advisor_result):
if not self._request_approval(advisor_result):
return "Task cancelled by user"
# Phase 2: Execution
return self._run_execution_phase(user_message, effective_system, advisor_result)
24.5 Real-World Case: Automated Data Migration
def demo_data_migration():
"""Demonstrate the Advisor tool in a high-risk data migration task"""
execution_tools = [
{
"name": "list_database_tables",
"description": "List all tables in a database",
"input_schema": {
"type": "object",
"properties": {"database": {"type": "string"}},
"required": ["database"]
}
},
{
"name": "backup_table",
"description": "Backup a specified table to backup storage",
"input_schema": {
"type": "object",
"properties": {
"table_name": {"type": "string"},
"backup_name": {"type": "string"}
},
"required": ["table_name", "backup_name"]
}
},
{
"name": "migrate_data",
"description": "Migrate data from source table to target table (HIGH RISK)",
"input_schema": {
"type": "object",
"properties": {
"source_table": {"type": "string"},
"target_table": {"type": "string"},
"transformation_sql": {"type": "string"}
},
"required": ["source_table", "target_table"]
}
},
{
"name": "verify_migration",
"description": "Verify the integrity of migrated data",
"input_schema": {
"type": "object",
"properties": {
"source_table": {"type": "string"},
"target_table": {"type": "string"}
},
"required": ["source_table", "target_table"]
}
}
]
tool_functions = {
"list_database_tables": lambda database: {
"tables": ["users", "orders", "products", "analytics_old"]
},
"backup_table": lambda table_name, backup_name: {
"success": True,
"backup_path": f"/backups/{backup_name}.sql",
"row_count": 50000
},
"migrate_data": lambda source_table, target_table, transformation_sql=None: {
"success": True,
"migrated_rows": 50000,
"failed_rows": 0,
"duration_seconds": 12.5
},
"verify_migration": lambda source_table, target_table: {
"source_count": 50000,
"target_count": 50000,
"checksums_match": True
}
}
agent = AdvisorExecutorAgent(
execution_tools=execution_tools,
tool_functions=tool_functions,
require_approval_for=["high", "critical"]
)
result = agent.run(
user_message="""Migrate the 'analytics_old' table from the production database 'prod_db'
to a new 'analytics_v2' table. During migration, convert the timestamp field from
Unix timestamp format to ISO 8601 format.""",
system="""You are an AI assistant responsible for database operations.
For data migration tasks, you must be extra cautious to ensure data safety and integrity."""
)
print("\n=== Final Result ===")
print(result)
return result
24.6 Combining Advisor with Extended Thinking
The Advisor Tool pairs well with Extended Thinking for deeper planning:
class ThinkingAdvisorAgent(AdvisorExecutorAgent):
"""Enhanced agent using Extended Thinking in the Advisor phase"""
def _run_advisor_phase(self, user_message: str, system: str) -> dict:
messages = [{"role": "user", "content": user_message}]
response = self.client.messages.create(
model=self.model,
max_tokens=16000,
thinking={"type": "enabled", "budget_tokens": 8000},
system=system + "\n\nThink deeply about all aspects of the task, then call the advisor tool to output a structured plan.",
tools=self.advisor_tools,
tool_choice={"type": "tool", "name": "advisor"},
messages=messages
)
for block in response.content:
if block.type == "thinking":
print(f"\n[Deep thinking] {len(block.thinking)} chars")
print(f"Preview: {block.thinking[:200]}...")
elif block.type == "tool_use" and block.name == "advisor":
self.advisor_result = block.input
return block.input
return {}
24.7 Metacognitive Feedback Loop: Post-Execution Self-Review
The Advisor pattern can be extended into a complete metacognitive feedback loop:
post_execution_review_tool = {
"name": "post_execution_review",
"description": """Call this tool after task execution to perform self-assessment.
Evaluation covers: whether execution matched the plan, unexpected situations encountered,
and how to improve next time.""",
"input_schema": {
"type": "object",
"properties": {
"planned_steps": {"type": "integer", "description": "Number of planned steps"},
"actual_steps": {"type": "integer", "description": "Number of steps actually executed"},
"deviations": {
"type": "array",
"items": {"type": "string"},
"description": "Deviations between actual execution and plan"
},
"unexpected_issues": {
"type": "array",
"items": {"type": "string"},
"description": "Unexpected problems encountered during execution"
},
"task_success": {"type": "boolean"},
"lessons_learned": {
"type": "array",
"items": {"type": "string"},
"description": "Lessons learned from this execution"
},
"quality_score": {
"type": "integer",
"minimum": 1,
"maximum": 10,
"description": "Self-assessed quality score (1-10)"
}
},
"required": ["task_success", "quality_score"]
}
}
24.8 Decision Tree for Choosing the Advisor Pattern
Is the task clear and simple?
├── Yes → Execute directly, no Advisor needed
└── No ↓
Does the task have more than 3 steps?
├── No → Light Advisor (check dependencies only)
└── Yes ↓
Does the task involve irreversible operations?
├── No → Standard Advisor (planning + risk assessment)
└── Yes ↓
What is the cost of task failure?
├── Low (can redo) → Advisor + automatic execution
├── Medium (has cost) → Advisor + confirmation at critical steps
└── High (catastrophic) → Advisor + full human approval
def select_advisor_mode(task: str) -> str:
"""Select Advisor mode based on task characteristics"""
irreversible_keywords = [
"delete", "send", "submit", "pay", "migrate", "format", "drop", "remove"
]
complex_keywords = [
"step", "then", "next", "after", "finally", "workflow", "process", "pipeline"
]
task_lower = task.lower()
is_irreversible = any(kw in task_lower for kw in irreversible_keywords)
is_complex = any(kw in task_lower for kw in complex_keywords) or len(task) > 200
if not is_complex and not is_irreversible:
return "direct" # Execute directly
elif is_complex and not is_irreversible:
return "light_advisor" # Light planning
elif not is_complex and is_irreversible:
return "safety_advisor" # Safety confirmation
else:
return "full_advisor" # Full planning + approval
Summary
The Advisor Tool pattern transforms Claude's metacognitive capability into observable, auditable tool calls, providing a "plan first, execute second" safety guarantee for complex tasks. The core value lies in:
- Transparency: The execution plan is presented as structured data that humans can understand and review
- Controllability: The
recommendationfield and risk levels make human intervention points clearly visible - Safety: Irreversible operations must go through explicit planning and optional human approval
- Improvability: Through
post_execution_review, a metacognitive feedback loop is established
Combined with Extended Thinking, the Advisor Tool enables Claude to demonstrate its deepest thinking and most systematic planning on the highest-risk tasks.
With this chapter, Part 4 of the "Complete Guide to Claude" — Tool Use and Function Calling — is complete. The next part will explore multi-agent architectures, advanced prompt engineering techniques, and Claude's security and compliance practices in production environments.