distributed-agent-mesh 0.1.0
A framework for autonomous agents that collaborate via P2P communication
License: MIT
Requires Python: >=3.8
Keywords: agents, distributed, p2p, autonomous, collaboration, mesh, multi-agent, ai

# Distributed Agent Mesh (DAM)

🚀 **A revolutionary framework for autonomous agents that collaborate via P2P communication**

[![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

## 🎯 What is DAM?

Distributed Agent Mesh (DAM) is a **peer-to-peer framework** where autonomous AI agents:
- 🤖 **Decide autonomously** if they can contribute to tasks
- 🤝 **Self-organize** into teams without central control
- ⚡ **Execute in parallel** for maximum performance
- 🔄 **Collaborate directly** via P2P communication
- 📈 **Scale dynamically** by adding/removing agents anytime
- 💪 **Tolerate failures** - system continues if agents fail

## 🆚 DAM vs MCP (Model Context Protocol)

### Architecture Comparison

| Aspect | MCP | DAM |
|--------|-----|-----|
| **Architecture** | Client-Server | P2P Mesh |
| **Control** | Centralized (LLM) | Distributed (Agents) |
| **Decision Making** | LLM decides everything | Agents decide autonomously |
| **Execution** | Sequential tool calls | Parallel agent collaboration |
| **Scalability** | Limited by LLM | Add agents anytime |
| **Resilience** | Single point of failure | Fault tolerant |
| **Performance** | Sequential (1x baseline) | **2-5x faster** (parallel) |

### When to Use What?

**Use MCP when:**
- ✅ Simple tool integration
- ✅ Single-agent systems
- ✅ Well-defined sequential workflows
- ✅ Resource access (files, databases)

**Use DAM when:**
- ✅ Complex, multi-step tasks
- ✅ Tasks requiring parallelism
- ✅ Large-scale distributed systems
- ✅ Systems requiring high availability
- ✅ Dynamic, adaptive workflows

**💡 Best Approach: Hybrid**
- Use MCP for resource access (data, tools)
- Use DAM for agent coordination
- Get the best of both worlds!

## 🚀 Installation

```bash
pip install distributed-agent-mesh
```

Or install from source:

```bash
git clone https://github.com/yourusername/distributed-agent-mesh.git
cd distributed-agent-mesh
pip install -e .
```

## 📖 Quick Start

### Basic Example

```python
import asyncio
from distributed_agent_mesh import (
    DistributedAgentMesh,
    AutonomousAgent,
    AgentCapability,
    Task
)

# Define custom executor
async def research_executor(task):
    # Your custom logic here
    return {
        'status': 'completed',
        'result': f'Research completed: {task.description}'
    }

# Create mesh
mesh = DistributedAgentMesh()

# Create autonomous agent
agent = AutonomousAgent(
    agent_id="research_agent",
    capabilities=[
        AgentCapability("research", "Conduct research")
    ],
    executor=research_executor
)

# Register agent
mesh.register_agent(agent)

# Create task
task = Task(
    id="task_001",
    description="Research AI trends",
    requirements=["research"],
    priority=1
)

# Agents collaborate autonomously!
async def main():
    result = await mesh.solve_complex_task(task)
    print(f"Result: {result}")

asyncio.run(main())
```

## 🌟 Real-World Example: Document Analysis

```python
import asyncio
from distributed_agent_mesh import (
    DistributedAgentMesh,
    AutonomousAgent,
    AgentCapability,
    Task
)

# Create specialized agents
async def text_extractor(task):
    # Extract text from documents
    return {'text': '...', 'pages': 10}

async def sentiment_analyzer(task):
    # Analyze sentiment
    return {'sentiment': 'positive', 'confidence': 0.89}

async def summarizer(task):
    # Generate summary
    return {'summary': '...', 'key_points': [...]}

# Setup mesh
mesh = DistributedAgentMesh()

# Register agents with capabilities
mesh.register_agent(AutonomousAgent(
    "text_agent",
    [AgentCapability("text_extraction", "Extract text")],
    text_extractor
))

mesh.register_agent(AutonomousAgent(
    "sentiment_agent",
    [AgentCapability("sentiment_analysis", "Analyze sentiment")],
    sentiment_analyzer
))

mesh.register_agent(AutonomousAgent(
    "summary_agent",
    [AgentCapability("summarization", "Summarize")],
    summarizer
))

# Process documents in parallel
async def analyze_documents():
    tasks = [
        Task(
            id=f"doc_{i}",
            description=f"Document {i}",
            requirements=["text_extraction", "sentiment_analysis", "summarization"],
            priority=1
        )
        for i in range(10)
    ]
    
    # All documents processed in parallel!
    results = await asyncio.gather(*[
        mesh.solve_complex_task(task) for task in tasks
    ])
    
    return results

# Run
results = asyncio.run(analyze_documents())
```

## 🎯 Key Features

### 1. Autonomous Decision Making

Agents decide on their own if they can contribute:

```python
async def can_contribute(self, task: Task) -> bool:
    """Agent decides autonomously"""
    relevance = await self.analyze_relevance(task)
    availability = self.check_availability()
    capability = self.check_capabilities(task)
    
    return relevance > 0.7 and availability and capability
```

### 2. Self-Organizing Teams

No central controller - agents form teams automatically:

```python
# Mesh broadcasts task
await mesh.solve_complex_task(task)

# Agents autonomously:
# 1. Decide if they can help
# 2. Self-organize into optimal team
# 3. Collaborate peer-to-peer
# 4. Execute in parallel
```

### 3. Dynamic Scalability

Add/remove agents anytime without restart:

```python
# Start with 2 agents
mesh.register_agent(agent1)
mesh.register_agent(agent2)

# Process tasks
result1 = await mesh.solve_complex_task(task1)

# High load? Add more agents dynamically!
mesh.register_agent(agent3)
mesh.register_agent(agent4)

# System automatically uses new agents
result2 = await mesh.solve_complex_task(task2)
```

### 4. Fault Tolerance

System continues working even when agents fail:

```python
# 5 agents working
mesh.register_agent(agent1)
mesh.register_agent(agent2)
# ... agent5

# 2 agents fail mid-execution
mesh.unregister_agent("agent1")
mesh.unregister_agent("agent2")

# System continues with remaining 3 agents!
result = await mesh.solve_complex_task(task)  # Still completes
```

## 📊 Performance Benchmarks

Based on our POC comparison:

### Single Task Performance
- **MCP (Sequential)**: 2.40s
- **DAM (Parallel)**: 1.00s
- **🚀 Speedup**: 2.4x faster

### Multiple Tasks (3 tasks)
- **MCP (Sequential)**: 7.21s
- **DAM (Parallel)**: 2.33s
- **🚀 Speedup**: 3.1x faster

### Scalability Test
- **Add agents dynamically**: ✅ No restart needed
- **Fault tolerance**: ✅ Continues despite failures
- **Load balancing**: ✅ Automatic distribution

## 🏗️ Architecture

```
┌─────────────────────────────────────────────────┐
│         Distributed Agent Mesh (DAM)            │
│                                                 │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐ │
│  │ Agent 1  │◄──►│ Agent 2  │◄──►│ Agent 3  │ │
│  │          │    │          │    │          │ │
│  │ Research │    │ Analysis │    │   Viz    │ │
│  └──────────┘    └──────────┘    └──────────┘ │
│       ▲               ▲               ▲        │
│       │               │               │        │
│       └───────────────┴───────────────┘        │
│              P2P Network Layer                 │
│                                                 │
│  • Autonomous decision making                  │
│  • Self-organizing teams                       │
│  • Parallel execution                          │
│  • Fault tolerance                             │
└─────────────────────────────────────────────────┘
```

## 🛠️ Advanced Usage

### Custom Agent Executor

```python
async def advanced_executor(task: Task) -> Dict[str, Any]:
    """Custom logic with LLM, tools, etc."""
    
    # Use LLM
    llm_response = await call_llm(task.description)
    
    # Use tools (via MCP or direct)
    tool_result = await use_tool("search", {"query": task.description})
    
    # Process data
    processed = process_results(llm_response, tool_result)
    
    return {
        'status': 'completed',
        'result': processed,
        'metadata': {'time': time.time()}
    }

agent = AutonomousAgent(
    "advanced_agent",
    [AgentCapability("advanced_task", "Advanced processing")],
    advanced_executor
)
```

### Monitoring and Stats

```python
# Get mesh statistics
stats = mesh.get_mesh_stats()

print(f"Total Agents: {stats['total_agents']}")
print(f"Active Agents: {stats['active_agents']}")
print(f"Completed Tasks: {stats['completed_tasks']}")
print(f"Peer Connections: {stats['total_peers']}")

# Individual agent stats
for agent_stat in stats['agent_stats']:
    print(f"{agent_stat['agent_id']}:")
    print(f"  Completed: {agent_stat['completed_tasks']}")
    print(f"  Performance: {agent_stat['performance_score']}")
```

## 🔬 Research & POC

We've conducted comprehensive research comparing DAM with MCP:

**Run the POC yourself:**

```bash
cd dam_research
python poc_comparison.py
```

**Results:**
- ✅ DAM is 2-3x faster than MCP
- ✅ True parallelism vs sequential execution
- ✅ Fault tolerance and resilience
- ✅ Dynamic scalability without restart

## 📚 Examples

Check out the `examples/` directory for:

1. **Basic Usage** (`examples/basic.py`)
2. **Document Analysis** (`examples/document_analysis.py`)
3. **Real-time Streaming** (`examples/streaming.py`)
4. **MCP vs DAM Comparison** (`examples/poc_comparison.py`)

## 🤝 Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

## 📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

## 🙏 Acknowledgments

- Inspired by the concept of distributed autonomous agents
- Built on Python's asyncio for high-performance async operations
- Zero external dependencies for maximum portability

## 📞 Support

- **Issues**: [GitHub Issues](https://github.com/yourusername/distributed-agent-mesh/issues)
- **Discussions**: [GitHub Discussions](https://github.com/yourusername/distributed-agent-mesh/discussions)
- **Email**: your.email@example.com

---

⭐ **Star us on GitHub if you find this useful!**

Made with ❤️ for the AI agent community