python-sdk/pkg/hanzo-mcp/examples/agent_grinding_demo.py

#!/usr/bin/env python3
"""Agent Grinding Pattern Demo.

This demonstrates the pattern where agents can launch with all normal tools,
read/research/study then decide to edit or not. You can fuzzy match things,
and just throw matches at agent grinder, let them grind through each until
it's all resolved.

The idea is that agents autonomously:
1. Search for relevant files/patterns
2. Study the code to understand context
3. Decide if changes are needed
4. Make precise edits if necessary
5. Move to the next match

This is efficient because:
- Agents work in parallel on different files
- Each agent has full context and tools
- They can skip files that don't need changes
- They make surgical edits when needed
"""

import asyncio
import json


# Simulated swarm tool usage
async def run_agent_grinding_demo():
    """Run the agent grinding pattern demo."""

    print("=== Agent Grinding Pattern Demo ===\n")

    # Example 1: Fix all TODO comments in a codebase
    print("Example 1: Resolving TODO comments across codebase")
    print("-" * 50)

    todo_grinding_config = {
        "query": "Resolve all TODO comments in the codebase",
        "agents": [
            {
                "id": "todo_finder",
                "query": "Find all TODO comments using grep and organize them by file",
                "role": "finder",
            },
            {
                "id": "todo_analyzer_1",
                "query": "Analyze TODOs in file group 1 and determine which need fixing",
                "role": "analyzer",
                "receives_from": ["todo_finder"],
            },
            {
                "id": "todo_analyzer_2",
                "query": "Analyze TODOs in file group 2 and determine which need fixing",
                "role": "analyzer",
                "receives_from": ["todo_finder"],
            },
            {
                "id": "todo_fixer_1",
                "query": "Fix the TODOs that need fixing in group 1. Read each file, understand context, make precise edits",
                "role": "fixer",
                "receives_from": ["todo_analyzer_1"],
            },
            {
                "id": "todo_fixer_2",
                "query": "Fix the TODOs that need fixing in group 2. Read each file, understand context, make precise edits",
                "role": "fixer",
                "receives_from": ["todo_analyzer_2"],
            },
            {
                "id": "reviewer",
                "query": "Review all changes made and ensure they're correct",
                "role": "reviewer",
                "receives_from": ["todo_fixer_1", "todo_fixer_2"],
            },
        ],
    }

    print(json.dumps(todo_grinding_config, indent=2))
    print("\nThis would launch 6 agents working in parallel to grind through TODOs\n")

    # Example 2: Update all deprecated API usage
    print("\nExample 2: Update deprecated API usage")
    print("-" * 50)

    api_update_config = {
        "query": "Update all deprecated API calls to new version",
        "agents": [
            {
                "id": "api_scanner",
                "query": "Search for all uses of old_api.* pattern in the codebase",
                "role": "scanner",
            },
            {
                "id": "api_updater_1",
                "query": "For files 1-10: Read file, understand usage context, update to new_api if appropriate",
                "role": "updater",
                "receives_from": ["api_scanner"],
            },
            {
                "id": "api_updater_2",
                "query": "For files 11-20: Read file, understand usage context, update to new_api if appropriate",
                "role": "updater",
                "receives_from": ["api_scanner"],
            },
            {
                "id": "api_updater_3",
                "query": "For files 21+: Read file, understand usage context, update to new_api if appropriate",
                "role": "updater",
                "receives_from": ["api_scanner"],
            },
            {
                "id": "test_runner",
                "query": "Run tests on all updated files to ensure changes work",
                "role": "tester",
                "receives_from": ["api_updater_1", "api_updater_2", "api_updater_3"],
            },
        ],
    }

    print(json.dumps(api_update_config, indent=2))
    print("\nAgents work in parallel, each handling a subset of files\n")

    # Example 3: Add type hints to untyped functions
    print("\nExample 3: Add type hints to Python functions")
    print("-" * 50)

    type_hint_config = {
        "query": "Add type hints to all Python functions missing them",
        "network_type": "pipeline",
        "agents": [
            {
                "id": "type_finder",
                "query": "Use grep_ast to find all Python functions without type hints",
                "role": "finder",
            },
            {
                "id": "type_inferrer",
                "query": "For each function, analyze usage and infer appropriate types",
                "role": "analyzer",
                "receives_from": ["type_finder"],
            },
            {
                "id": "type_adder",
                "query": "Add the inferred type hints to each function signature",
                "role": "editor",
                "receives_from": ["type_inferrer"],
            },
            {
                "id": "mypy_checker",
                "query": "Run mypy on modified files to verify type correctness",
                "role": "validator",
                "receives_from": ["type_adder"],
            },
        ],
    }

    print(json.dumps(type_hint_config, indent=2))
    print("\nPipeline pattern: each stage processes all items before passing to next\n")

    # Example 4: Refactor duplicated code
    print("\nExample 4: Refactor duplicated code patterns")
    print("-" * 50)

    refactor_config = {
        "query": "Find and refactor duplicated code patterns",
        "consensus_mode": True,
        "agents": [
            {
                "id": "dup_finder",
                "query": "Find similar code patterns that might be duplicated",
                "role": "finder",
            },
            {
                "id": "refactor_designer",
                "query": "Design refactoring approach for each duplication",
                "role": "architect",
                "receives_from": ["dup_finder"],
            },
            {
                "id": "consensus_group",
                "query": "Discuss and agree on best refactoring approach",
                "role": "consensus",
                "model": "claude-3-5-sonnet-20241022",
                "participants": 3,
                "receives_from": ["refactor_designer"],
            },
            {
                "id": "refactorer",
                "query": "Implement the agreed refactoring",
                "role": "implementer",
                "receives_from": ["consensus_group"],
            },
        ],
    }

    print(json.dumps(refactor_config, indent=2))
    print("\nConsensus mode: multiple agents discuss before making changes\n")

    # Show the key benefits
    print("\n=== Key Benefits of Agent Grinding ===")
    print(
        "1. Parallel Processing: Multiple agents work on different files simultaneously"
    )
    print(
        "2. Full Context: Each agent has all tools - can read, search, understand before editing"
    )
    print("3. Smart Filtering: Agents skip files that don't need changes")
    print("4. Precise Edits: Agents make surgical changes based on understanding")
    print("5. Scalable: Add more agents to handle larger codebases")
    print(
        "6. Flexible: Different patterns (pipeline, parallel, consensus) for different tasks"
    )

    # Show example swarm command
    print("\n=== Example Swarm Command ===")
    print("""
swarm_tool(
    query="Fix all FIXME comments in the codebase",
    agents=[
        {"id": "finder", "query": "grep for all FIXME comments"},
        {"id": "grinder1", "query": "Fix FIXMEs in src/", "receives_from": ["finder"]},
        {"id": "grinder2", "query": "Fix FIXMEs in tests/", "receives_from": ["finder"]},
        {"id": "grinder3", "query": "Fix FIXMEs in docs/", "receives_from": ["finder"]},
        {"id": "reviewer", "query": "Review all fixes", "receives_from": ["grinder1", "grinder2", "grinder3"]}
    ]
)
""")

    print("\nThe agents will:")
    print("- finder: Search and categorize all FIXMEs")
    print("- grinders: Work in parallel on different directories")
    print(
        "- Each grinder: Read file → Understand context → Fix if needed → Move to next"
    )
    print("- reviewer: Ensure all fixes are appropriate")


if __name__ == "__main__":
    asyncio.run(run_agent_grinding_demo())