Files
ajarbot/examples/sub_agent_example.py
Jordan Ramos 50cf7165cb Add sub-agent orchestration, MCP tools, and critical bug fixes
Major Features:
- Sub-agent orchestration system with dynamic specialist spawning
  * spawn_sub_agent(): Create specialists with custom prompts
  * delegate(): Convenience method for task delegation
  * Cached specialists for reuse
  * Separate conversation histories and focused context

- MCP (Model Context Protocol) tool integration
  * Zettelkasten: fleeting_note, daily_note, permanent_note, literature_note
  * Search: search_vault (hybrid search), search_by_tags
  * Web: web_fetch for real-time data
  * Zero-cost file/system operations on Pro subscription

Critical Bug Fixes:
- Fixed max tool iterations (15 → 30, configurable)
- Fixed max_tokens error in Agent SDK query() call
- Fixed MCP tool routing in execute_tool()
  * Routes zettelkasten + web tools to async handlers
  * Prevents "Unknown tool" errors

Documentation:
- SUB_AGENTS.md: Complete guide to sub-agent system
- MCP_MIGRATION.md: Agent SDK migration details
- SOUL.example.md: Sanitized bot identity template
- scheduled_tasks.example.yaml: Sanitized task config template

Security:
- Added obsidian vault to .gitignore
- Protected SOUL.md and MEMORY.md (personal configs)
- Sanitized example configs with placeholders

Dependencies:
- Added beautifulsoup4, httpx, lxml for web scraping
- Updated requirements.txt

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
2026-02-16 07:43:31 -07:00

174 lines
5.2 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Example: Using Sub-Agent Orchestration
This example demonstrates how to use the sub-agent system to delegate
specialized tasks to focused agents.
"""
import sys
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from agent import Agent
def example_1_manual_spawning():
"""Example 1: Manually spawn and use a specialist."""
print("=== Example 1: Manual Spawning ===\n")
# Create main agent
agent = Agent(provider="claude")
# Spawn a zettelkasten specialist
zettel_specialist = agent.spawn_sub_agent(
specialist_prompt="""You are a zettelkasten expert. Your ONLY job is:
- Process fleeting notes into permanent notes
- Find semantic connections using hybrid search
- Create wiki-style links between related concepts
Stay focused on knowledge management. Be concise.""",
agent_id="zettelkasten_processor" # Cached for reuse
)
# Use the specialist
result = zettel_specialist.chat(
"Search for all fleeting notes tagged 'AI' and show me what you find.",
username="jordan"
)
print(f"Specialist Response:\n{result}\n")
# Reuse the cached specialist
result2 = zettel_specialist.chat(
"Now create a permanent note summarizing key AI concepts.",
username="jordan"
)
print(f"Second Response:\n{result2}\n")
def example_2_delegation():
"""Example 2: One-off delegation (convenience method)."""
print("=== Example 2: Delegation ===\n")
agent = Agent(provider="claude")
# One-off delegation (specialist not cached)
result = agent.delegate(
task="List all files in the memory_workspace/obsidian directory",
specialist_prompt="""You are a file system expert. Your job is to:
- Navigate directories efficiently
- Provide clear, organized file listings
Be concise and focused.""",
username="jordan"
)
print(f"Delegation Result:\n{result}\n")
def example_3_cached_delegation():
"""Example 3: Cached delegation (reuse specialist)."""
print("=== Example 3: Cached Delegation ===\n")
agent = Agent(provider="claude")
# First call: Creates and caches the specialist
result1 = agent.delegate(
task="Search the zettelkasten vault for notes about 'architecture'",
specialist_prompt="""You are a zettelkasten search expert. Your job is:
- Use hybrid search to find relevant notes
- Summarize key findings concisely
Stay focused on search and retrieval.""",
username="jordan",
agent_id="zettel_search" # This specialist will be cached
)
print(f"First Search:\n{result1}\n")
# Second call: Reuses the cached specialist
result2 = agent.delegate(
task="Now search for notes about 'design patterns'",
specialist_prompt="(ignored - using cached specialist)",
username="jordan",
agent_id="zettel_search" # Same ID = reuse cached specialist
)
print(f"Second Search:\n{result2}\n")
def example_4_multiple_specialists():
"""Example 4: Use multiple specialists for different tasks."""
print("=== Example 4: Multiple Specialists ===\n")
agent = Agent(provider="claude")
# Email specialist
email_result = agent.delegate(
task="Check if there are any unread emails in the last 24 hours",
specialist_prompt="""You are an email analyst. Your job is:
- Search and filter emails efficiently
- Summarize key information concisely
Focus on email intelligence.""",
username="jordan",
agent_id="email_analyst"
)
print(f"Email Analysis:\n{email_result}\n")
# Calendar specialist
calendar_result = agent.delegate(
task="Show me my calendar events for the next 3 days",
specialist_prompt="""You are a calendar expert. Your job is:
- Retrieve calendar events efficiently
- Present schedules clearly
Focus on time management.""",
username="jordan",
agent_id="calendar_manager"
)
print(f"Calendar Review:\n{calendar_result}\n")
def example_5_isolated_memory():
"""Example 5: Create specialist with isolated memory."""
print("=== Example 5: Isolated Memory ===\n")
agent = Agent(provider="claude")
# Specialist with its own memory workspace
specialist = agent.spawn_sub_agent(
specialist_prompt="You are a research assistant. Focus on gathering information.",
agent_id="researcher",
share_memory=False # Isolated workspace
)
# This specialist's memory is stored in:
# memory_workspace/sub_agents/researcher/
result = specialist.chat(
"Research the concept of 'emergence' and save findings.",
username="jordan"
)
print(f"Research Result:\n{result}\n")
if __name__ == "__main__":
# Run examples
# Uncomment the examples you want to try:
# example_1_manual_spawning()
# example_2_delegation()
# example_3_cached_delegation()
# example_4_multiple_specialists()
# example_5_isolated_memory()
print("\n Uncomment the examples you want to run in the __main__ block")
print(" Note: Some examples require Google OAuth setup and active API keys")