## Introduction to Claude Code Subagents

Welcome to your comprehensive journey into mastering Claude Code subagents! This course will take you from understanding the basics to building a fully automated book-writing system.

What You'll Master: By the end of this course, you'll be able to create sophisticated multi-agent systems that can handle complex tasks like writing, editing, and publishing entire books automatically.

### What Are Subagents?

Subagents are specialized AI assistants that can be invoked to handle specific types of tasks. They enable more efficient problem-solving by providing task-specific configurations with customized system prompts, tools, and a separate context window.

#### 🎯 Task Specialization

Each subagent focuses on a specific domain, allowing for deeper expertise and better results

#### 🔧 Custom Configuration

Tailored system prompts and tools for each agent's specific responsibilities

#### 📊 Parallel Processing

Multiple agents can work simultaneously on different aspects of a problem

### Setting Up Your Environment

`# Install Claude Code CLI npm install -g claude-code # Verify installation claude-code --version # Initialize a new project claude-code init my-subagent-project cd my-subagent-project # Configure API credentials export ANTHROPIC_API_KEY="your-api-key-here" # Create your first agent configuration claude-code agent create my-first-agent`

### Project Structure

📁 my-subagent-project/
📁 .claude/
📁 agents/
📄 research-agent.md
📄 writer-agent.md
📄 editor-agent.md
📄 claude.json
📄 package.json

#### Try Your First Command

Click the button below to simulate creating your first subagent:

## Understanding Subagents

Let's dive deep into what makes subagents powerful and how they differ from regular Claude interactions.

### Core Concepts

#### Subagent Anatomy Core

Every subagent consists of:

- Name: Unique identifier for the agent
- Description: When and how to use this agent
- System Prompt: Specialized instructions defining the agent's expertise
- Tools: Specific capabilities available to the agent
- Examples: Usage patterns with commentary

### Real-World Agent Examples

#### Python Backend Engineer Agent

`--- name: python-backend-engineer description: Use this agent when you need to develop, refactor, or optimize Python backend systems using modern tooling like uv. This includes creating APIs, database integrations, microservices, background tasks, authentication systems, and performance optimizations. color: green --- You are a Senior Python Backend Engineer with deep expertise in modern Python development, specializing in building scalable, maintainable backend systems using cutting-edge tools like uv for dependency management and project setup. Your core responsibilities: - Design and implement robust backend architectures following SOLID principles - Write clean, modular, well-documented Python code with comprehensive type hints - Leverage uv for efficient dependency management and virtual environments - Create RESTful APIs and GraphQL endpoints with proper validation - Design efficient database schemas and implement optimized queries - Implement authentication, authorization, and security best practices - Write comprehensive unit and integration tests using pytest - Optimize performance through profiling, caching, and async programming`

#### UI Engineer Agent

`--- name: ui-engineer description: Use this agent when you need to create, modify, or review frontend code, UI components, or user interfaces. color: purple --- You are an expert UI engineer with deep expertise in modern frontend development, specializing in creating clean, maintainable, and highly readable code that seamlessly integrates with any backend system. Your Expertise Areas: - Modern JavaScript/TypeScript with latest ES features - React, Vue, Angular, and contemporary frameworks - CSS-in-JS, Tailwind CSS, and modern styling approaches - Responsive design and mobile-first development - Component-driven architecture and design systems - State management patterns (Redux, Zustand, Context API) - Performance optimization and bundle analysis - Accessibility (WCAG) compliance and inclusive design`

#### Senior Code Reviewer Agent

`--- name: senior-code-reviewer description: Use this agent when you need comprehensive code review from a senior fullstack developer perspective. color: blue --- You are a Senior Fullstack Code Reviewer, an expert software architect with 15+ years of experience across frontend, backend, database, and DevOps domains. Core Responsibilities: - Conduct thorough code reviews with senior-level expertise - Analyze code for security vulnerabilities and performance bottlenecks - Evaluate architectural decisions and suggest improvements - Ensure adherence to coding standards and best practices - Identify potential bugs, edge cases, and error handling gaps - Assess test coverage and quality - Review database queries, API designs, and system integrations`

### Agent Invocation

`// How to invoke agents in Claude Code // Method 1: Direct invocation /agents python-backend-engineer // Method 2: With context /agents ui-engineer "Create a responsive navigation component" // Method 3: Chain multiple agents /agents research-agent "Gather info on React patterns" /agents ui-engineer "Implement the patterns we found" /agents senior-code-reviewer "Review the implementation"`

Best Practices for Agent Design:

- Keep agents focused on a single domain or responsibility
- Provide clear, detailed system prompts
- Include concrete examples with commentary
- Define when to use vs when not to use the agent
- Specify the tools and capabilities needed

#### Explore Agent Capabilities

## Creating Your First Subagent

Let's build a complete research subagent from scratch that can gather and analyze information on any topic.

### Step 1: Define the Agent Configuration

`--- name: research-specialist description: Use this agent when you need comprehensive research on any topic, including gathering information, analyzing sources, fact-checking, and synthesizing findings into actionable insights. color: cyan --- You are a Research Specialist with expertise in information gathering, analysis, and synthesis across multiple domains. You excel at finding reliable sources, evaluating credibility, and presenting findings in a clear, structured manner. Core Capabilities: - Comprehensive research across academic, industry, and popular sources - Critical evaluation of source credibility and bias - Fact-checking and cross-referencing information - Synthesis of complex information into clear insights - Identification of knowledge gaps and research opportunities - Trend analysis and pattern recognition - Creation of research summaries and reports Research Process: 1. Define Research Scope: Clarify objectives and constraints 2. Source Identification: Find relevant, credible sources 3. Information Gathering: Collect comprehensive data 4. Analysis & Validation: Verify facts and evaluate quality 5. Synthesis: Combine findings into coherent insights 6. Presentation: Structure findings for the target audience When conducting research: - Prioritize primary sources over secondary when possible - Note any conflicting information or controversies - Identify potential biases in sources - Provide confidence levels for findings - Suggest areas for further investigation - Include relevant citations and references`

### Step 2: Create Supporting Code Structure

`# research_agent.py from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime import json import asyncio from enum import Enum class ResearchDepth(Enum): QUICK = "quick" STANDARD = "standard" COMPREHENSIVE = "comprehensive" EXHAUSTIVE = "exhaustive" class SourceType(Enum): ACADEMIC = "academic" INDUSTRY = "industry" NEWS = "news" GOVERNMENT = "government" EXPERT = "expert" @dataclass class ResearchSource: """Represents a research source with metadata""" title: str url: Optional[str] type: SourceType credibility_score: float date_published: Optional[datetime] key_points: List[str] bias_assessment: str @dataclass class ResearchFinding: """Individual research finding with supporting evidence""" claim: str evidence: List[str] sources: List[ResearchSource] confidence_level: float contradictions: List[str] class ResearchAgent: """Advanced research agent with comprehensive capabilities""" def __init__(self, api_key: str): self.api_key = api_key self.research_cache = {} self.source_evaluator = SourceEvaluator() self.fact_checker = FactChecker() async def research( self, topic: str, depth: ResearchDepth = ResearchDepth.STANDARD, source_types: Optional[List[SourceType]] = None, time_range: Optional[tuple] = None ) -> Dict[str, Any]: """ Conduct comprehensive research on a topic Args: topic: The research topic depth: Level of research depth source_types: Types of sources to include time_range: Date range for sources (start, end) Returns: Comprehensive research report """ print(f"🔍 Starting {depth.value} research on: {topic}") # Check cache first cache_key = self._generate_cache_key(topic, depth) if cache_key in self.research_cache: print("📦 Returning cached results") return self.research_cache[cache_key] # Define research scope scope = self._define_scope(topic, depth, source_types) # Gather sources sources = await self._gather_sources(scope, time_range) # Analyze and validate findings = await self._analyze_sources(sources) # Check facts validated_findings = await self._validate_findings(findings) # Synthesize results synthesis = self._synthesize_findings(validated_findings) # Generate report report = self._generate_report( topic=topic, scope=scope, sources=sources, findings=validated_findings, synthesis=synthesis ) # Cache results self.research_cache[cache_key] = report return report def _define_scope( self, topic: str, depth: ResearchDepth, source_types: Optional[List[SourceType]] ) -> Dict[str, Any]: """Define the research scope based on parameters""" scope = { 'topic': topic, 'depth': depth, 'source_types': source_types or list(SourceType), 'subtopics': self._identify_subtopics(topic, depth), 'keywords': self._generate_keywords(topic), 'questions': self._generate_research_questions(topic, depth) } return scope def _identify_subtopics(self, topic: str, depth: ResearchDepth) -> List[str]: """Identify relevant subtopics based on depth""" base_subtopics = [ f"{topic} fundamentals", f"{topic} applications", f"{topic} challenges" ] if depth in [ResearchDepth.COMPREHENSIVE, ResearchDepth.EXHAUSTIVE]: base_subtopics.extend([ f"{topic} history", f"{topic} future trends", f"{topic} expert opinions", f"{topic} case studies" ]) if depth == ResearchDepth.EXHAUSTIVE: base_subtopics.extend([ f"{topic} methodologies", f"{topic} comparative analysis", f"{topic} global perspectives", f"{topic} economic impact" ]) return base_subtopics async def _gather_sources( self, scope: Dict[str, Any], time_range: Optional[tuple] ) -> List[ResearchSource]: """Gather sources based on scope""" sources = [] for source_type in scope['source_types']: type_sources = await self._fetch_sources_by_type( scope['topic'], source_type, scope['keywords'], time_range ) sources.extend(type_sources) # Evaluate source credibility for source in sources: source.credibility_score = self.source_evaluator.evaluate(source) # Sort by credibility sources.sort(key=lambda x: x.credibility_score, reverse=True) return sources async def _analyze_sources( self, sources: List[ResearchSource] ) -> List[ResearchFinding]: """Analyze sources and extract findings""" findings = [] # Group sources by topic topic_groups = self._group_sources_by_topic(sources) for topic, topic_sources in topic_groups.items(): # Extract claims from sources claims = self._extract_claims(topic_sources) for claim in claims: finding = ResearchFinding( claim=claim['text'], evidence=claim['evidence'], sources=claim['sources'], confidence_level=self._calculate_confidence(claim), contradictions=self._find_contradictions(claim, claims) ) findings.append(finding) return findings async def _validate_findings( self, findings: List[ResearchFinding] ) -> List[ResearchFinding]: """Validate findings through fact-checking""" validated = [] for finding in findings: # Fact-check the claim is_valid = await self.fact_checker.check(finding.claim, finding.evidence) if is_valid: validated.append(finding) else: # Adjust confidence level for disputed claims finding.confidence_level *= 0.5 validated.append(finding) return validated def _synthesize_findings( self, findings: List[ResearchFinding] ) -> Dict[str, Any]: """Synthesize findings into coherent insights""" synthesis = { 'key_insights': self._extract_key_insights(findings), 'consensus_points': self._identify_consensus(findings), 'controversial_areas': self._identify_controversies(findings), 'knowledge_gaps': self._identify_gaps(findings), 'confidence_assessment': self._assess_overall_confidence(findings), 'recommendations': self._generate_recommendations(findings) } return synthesis def _generate_report( self, topic: str, scope: Dict[str, Any], sources: List[ResearchSource], findings: List[ResearchFinding], synthesis: Dict[str, Any] ) -> Dict[str, Any]: """Generate comprehensive research report""" report = { 'metadata': { 'topic': topic, 'timestamp': datetime.now().isoformat(), 'depth': scope['depth'].value, 'source_count': len(sources), 'finding_count': len(findings) }, 'executive_summary': self._create_executive_summary(synthesis), 'detailed_findings': findings, 'source_analysis': { 'total_sources': len(sources), 'by_type': self._count_sources_by_type(sources), 'credibility_distribution': self._analyze_credibility(sources), 'date_range': self._get_date_range(sources) }, 'synthesis': synthesis, 'bibliography': self._format_bibliography(sources), 'appendices': { 'methodology': self._describe_methodology(scope), 'limitations': self._identify_limitations(scope, sources), 'further_research': synthesis['knowledge_gaps'] } } return report def _generate_cache_key(self, topic: str, depth: ResearchDepth) -> str: """Generate cache key for research results""" return f"{topic.lower().replace(' ', '_')}_{depth.value}" class SourceEvaluator: """Evaluates credibility of research sources""" def evaluate(self, source: ResearchSource) -> float: """Evaluate source credibility on 0-1 scale""" score = 0.5 # Base score # Adjust based on source type type_scores = { SourceType.ACADEMIC: 0.9, SourceType.GOVERNMENT: 0.85, SourceType.EXPERT: 0.8, SourceType.INDUSTRY: 0.7, SourceType.NEWS: 0.6 } score = type_scores.get(source.type, 0.5) # Adjust for recency if source.date_published: days_old = (datetime.now() - source.date_published).days if days_old < 30: score += 0.1 elif days_old > 365: score -= 0.1 return min(max(score, 0), 1) # Clamp to 0-1 class FactChecker: """Validates claims through fact-checking""" async def check(self, claim: str, evidence: List[str]) -> bool: """Check if a claim is supported by evidence""" # Simplified fact-checking logic # In production, this would use external APIs or databases if len(evidence) >= 2: return True return False # Example usage async def main(): agent = ResearchAgent(api_key="your-api-key") # Conduct research results = await agent.research( topic="Quantum Computing Applications", depth=ResearchDepth.COMPREHENSIVE, source_types=[SourceType.ACADEMIC, SourceType.INDUSTRY], time_range=(datetime(2020, 1, 1), datetime.now()) ) # Print executive summary print("\n📊 Executive Summary:") print(results['executive_summary']) # Print key insights print("\n💡 Key Insights:") for insight in results['synthesis']['key_insights']: print(f" • {insight}") if __name__ == "__main__": asyncio.run(main())`

#### Test Your Research Agent

## Agent Communication & Coordination

Learn how to make agents work together through effective communication patterns and coordination strategies.

### Communication Patterns

Request
→
Process
→
Response
→
Validate

### Message Bus Implementation

`# agent_communication.py import asyncio from typing import Dict, Any, List, Optional, Callable from dataclasses import dataclass from datetime import datetime import json from enum import Enum import uuid class MessageType(Enum): REQUEST = "request" RESPONSE = "response" BROADCAST = "broadcast" ERROR = "error" STATUS = "status" @dataclass class Message: """Inter-agent message structure""" id: str sender: str recipient: str type: MessageType content: Any timestamp: datetime correlation_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { 'id': self.id, 'sender': self.sender, 'recipient': self.recipient, 'type': self.type.value, 'content': self.content, 'timestamp': self.timestamp.isoformat(), 'correlation_id': self.correlation_id } class AgentCommunicationBus: """Central communication hub for agent coordination""" def __init__(self): self.agents: Dict[str, 'BaseAgent'] = {} self.channels: Dict[str, List[str]] = {} self.message_queue: asyncio.Queue = asyncio.Queue() self.message_history: List[Message] = [] self.handlers: Dict[str, List[Callable]] = {} self.running = False def register_agent(self, agent: 'BaseAgent'): """Register an agent with the communication bus""" self.agents[agent.id] = agent print(f"✅ Registered agent: {agent.id}") def create_channel(self, name: str, participants: List[str]): """Create a communication channel for specific agents""" self.channels[name] = participants print(f"📢 Created channel: {name} with {len(participants)} participants") async def send_message( self, sender: str, recipient: str, content: Any, message_type: MessageType = MessageType.REQUEST, correlation_id: Optional[str] = None ) -> str: """Send a message from one agent to another""" message = Message( id=str(uuid.uuid4()), sender=sender, recipient=recipient, type=message_type, content=content, timestamp=datetime.now(), correlation_id=correlation_id ) await self.message_queue.put(message) self.message_history.append(message) return message.id async def broadcast( self, sender: str, channel: str, content: Any ): """Broadcast a message to all agents in a channel""" if channel not in self.channels: raise ValueError(f"Channel {channel} does not exist") for recipient in self.channels[channel]: if recipient != sender: await self.send_message( sender=sender, recipient=recipient, content=content, message_type=MessageType.BROADCAST ) async def request_response( self, sender: str, recipient: str, content: Any, timeout: float = 30.0 ) -> Optional[Any]: """Send a request and wait for response""" request_id = await self.send_message( sender=sender, recipient=recipient, content=content, message_type=MessageType.REQUEST ) # Wait for response with matching correlation_id start_time = asyncio.get_event_loop().time() while asyncio.get_event_loop().time() - start_time < timeout: # Check message history for response for msg in reversed(self.message_history): if (msg.correlation_id == request_id and msg.type == MessageType.RESPONSE and msg.sender == recipient): return msg.content await asyncio.sleep(0.1) return None # Timeout async def process_messages(self): """Main message processing loop""" self.running = True print("🚀 Communication bus started") while self.running: try: # Get message from queue with timeout message = await asyncio.wait_for( self.message_queue.get(), timeout=1.0 ) # Deliver to recipient if message.recipient in self.agents: agent = self.agents[message.recipient] await agent.receive_message(message) # Trigger any registered handlers await self._trigger_handlers(message) except asyncio.TimeoutError: continue except Exception as e: print(f"❌ Error processing message: {e}") async def _trigger_handlers(self, message: Message): """Trigger registered message handlers""" key = f"{message.type.value}:{message.recipient}" if key in self.handlers: for handler in self.handlers[key]: try: await handler(message) except Exception as e: print(f"❌ Handler error: {e}") def register_handler( self, agent_id: str, message_type: MessageType, handler: Callable ): """Register a message handler for specific message types""" key = f"{message_type.value}:{agent_id}" if key not in self.handlers: self.handlers[key] = [] self.handlers[key].append(handler) def get_message_history( self, agent_id: Optional[str] = None, message_type: Optional[MessageType] = None, limit: int = 100 ) -> List[Message]: """Get message history with optional filters""" history = self.message_history if agent_id: history = [ m for m in history if m.sender == agent_id or m.recipient == agent_id ] if message_type: history = [m for m in history if m.type == message_type] return history[-limit:] def stop(self): """Stop the communication bus""" self.running = False print("🛑 Communication bus stopped") class BaseAgent: """Base class for all agents with communication capabilities""" def __init__(self, agent_id: str, name: str): self.id = agent_id self.name = name self.message_buffer: List[Message] = [] self.communication_bus: Optional[AgentCommunicationBus] = None def connect_to_bus(self, bus: AgentCommunicationBus): """Connect agent to communication bus""" self.communication_bus = bus bus.register_agent(self) async def receive_message(self, message: Message): """Receive and process incoming message""" self.message_buffer.append(message) # Process based on message type if message.type == MessageType.REQUEST: response = await self.handle_request(message.content) await self.send_response(message.sender, response, message.id) elif message.type == MessageType.BROADCAST: await self.handle_broadcast(message.content) elif message.type == MessageType.STATUS: await self.handle_status(message.content) async def send_message( self, recipient: str, content: Any, message_type: MessageType = MessageType.REQUEST ): """Send a message to another agent""" if not self.communication_bus: raise RuntimeError("Agent not connected to communication bus") await self.communication_bus.send_message( sender=self.id, recipient=recipient, content=content, message_type=message_type ) async def send_response( self, recipient: str, content: Any, correlation_id: str ): """Send a response to a request""" if not self.communication_bus: raise RuntimeError("Agent not connected to communication bus") await self.communication_bus.send_message( sender=self.id, recipient=recipient, content=content, message_type=MessageType.RESPONSE, correlation_id=correlation_id ) async def request_from_agent( self, agent_id: str, request: Any, timeout: float = 30.0 ) -> Optional[Any]: """Request information from another agent and wait for response""" if not self.communication_bus: raise RuntimeError("Agent not connected to communication bus") return await self.communication_bus.request_response( sender=self.id, recipient=agent_id, content=request, timeout=timeout ) async def broadcast_to_channel(self, channel: str, content: Any): """Broadcast message to a channel""" if not self.communication_bus: raise RuntimeError("Agent not connected to communication bus") await self.communication_bus.broadcast( sender=self.id, channel=channel, content=content ) # Abstract methods to be implemented by specific agents async def handle_request(self, content: Any) -> Any: """Handle incoming request""" raise NotImplementedError async def handle_broadcast(self, content: Any): """Handle broadcast message""" pass async def handle_status(self, content: Any): """Handle status update""" pass # Example: Specialized Agents class ResearchAgent(BaseAgent): """Agent specialized in research tasks""" def __init__(self): super().__init__("research_agent", "Research Specialist") self.knowledge_base = {} async def handle_request(self, content: Any) -> Any: """Process research requests""" if isinstance(content, dict) and 'topic' in content: # Simulate research research_result = { 'topic': content['topic'], 'findings': [ f"Key finding 1 about {content['topic']}", f"Key finding 2 about {content['topic']}", f"Key finding 3 about {content['topic']}" ], 'sources': ['Source A', 'Source B', 'Source C'], 'confidence': 0.85 } return research_result return {'error': 'Invalid research request'} class WriterAgent(BaseAgent): """Agent specialized in content writing""" def __init__(self): super().__init__("writer_agent", "Content Writer") self.writing_style = "professional" async def handle_request(self, content: Any) -> Any: """Process writing requests""" if isinstance(content, dict) and 'research' in content: # Generate content based on research written_content = { 'title': f"Article on {content['research']['topic']}", 'content': self._generate_content(content['research']), 'word_count': 500, 'style': self.writing_style } return written_content return {'error': 'Invalid writing request'} def _generate_content(self, research: Dict[str, Any]) -> str: """Generate content from research""" content = f"# {research['topic']}\n\n" content += "## Key Findings\n\n" for finding in research.get('findings', []): content += f"- {finding}\n" content += "\n## Conclusion\n\n" content += f"Based on the research with {research.get('confidence', 0):.0%} confidence..." return content class EditorAgent(BaseAgent): """Agent specialized in editing and quality control""" def __init__(self): super().__init__("editor_agent", "Content Editor") async def handle_request(self, content: Any) -> Any: """Process editing requests""" if isinstance(content, dict) and 'content' in content: # Edit and improve content edited_content = { 'original': content['content'], 'edited': self._edit_content(content['content']), 'suggestions': [ "Consider adding more specific examples", "Strengthen the conclusion", "Check for consistency in terminology" ], 'quality_score': 0.78 } return edited_content return {'error': 'Invalid editing request'} def _edit_content(self, content: str) -> str: """Apply edits to content""" # Simplified editing logic edited = content.replace(" ", " ") # Remove double spaces edited = edited.replace("\n\n\n", "\n\n") # Fix spacing return edited # Example: Multi-Agent Workflow async def run_content_creation_workflow(): """Demonstrate multi-agent content creation""" # Initialize communication bus bus = AgentCommunicationBus() # Initialize agents research_agent = ResearchAgent() writer_agent = WriterAgent() editor_agent = EditorAgent() # Connect agents to bus research_agent.connect_to_bus(bus) writer_agent.connect_to_bus(bus) editor_agent.connect_to_bus(bus) # Create a channel for content team bus.create_channel("content_team", [ research_agent.id, writer_agent.id, editor_agent.id ]) # Start communication bus bus_task = asyncio.create_task(bus.process_messages()) try: # Step 1: Research print("\n📚 Step 1: Conducting research...") research_result = await research_agent.handle_request({ 'topic': 'Artificial Intelligence in Healthcare' }) print(f"Research completed: {len(research_result['findings'])} findings") # Step 2: Writing print("\n✍️ Step 2: Writing content...") written_content = await writer_agent.request_from_agent( writer_agent.id, {'research': research_result} ) # Simulate the response since we're calling directly written_content = await writer_agent.handle_request({'research': research_result}) print(f"Content written: {written_content['word_count']} words") # Step 3: Editing print("\n📝 Step 3: Editing content...") edited_content = await editor_agent.handle_request({ 'content': written_content['content'] }) print(f"Content edited: Quality score {edited_content['quality_score']:.0%}") # Broadcast completion await research_agent.broadcast_to_channel( "content_team", { 'status': 'completed', 'topic': 'Artificial Intelligence in Healthcare', 'quality_score': edited_content['quality_score'] } ) print("\n✅ Workflow completed successfully!") # Print message history print("\n📨 Message History:") for msg in bus.get_message_history(limit=10): print(f" {msg.timestamp`