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