Traditional Multi-Agent Systems often require extensive coding to define agent behaviors, workflows, and interactions. This project takes a different approach: What if you could build an entire multi-agent system just by writing a configuration JSON file?

Instead of hard-coding agent logic, workflows, and system behaviors, this framework allows you to:

• Define agents and their capabilities through JSON configuration
• Specify workflow stages and transitions declaratively
• Configure error handling and recovery strategies
• Set up agent communication patterns
• Integrate with external services and APIs

All without writing a single line of agent implementation code.

1. Install the package:

pip install distributedapps-mas

2. Run the example test case:

# Clone the repository
git clone https://github.com/kenhuangus/mas.git
cd mas

# Install in development mode
pip install -e .

# Run the document processing example
python examples/document_processing/test_document_processing.py

You should see output like this:

Testing Document Processing Example
==================================

Running test case: Basic document
 Basic document: Passed
Input: Hello, Multi-Agent System!
Output: HELLO, MULTI-AGENT SYSTEM!

Running test case: Empty document
 Empty document: Passed
Input: 
Output: 

Running test case: Special characters
 Special characters: Passed
Input: Hello! @#$%^&*()_+
Output: HELLO! @#$%^&*()_+

 All tests passed!

1. Rapid Development

   • Create new agents by adding JSON configuration
   • Modify workflows without changing code
   • Test different agent configurations quickly

2. Reduced Complexity

   • No need to implement agent communication logic
   • Declarative workflow definitions
   • Built-in error handling and retries

3. Flexibility

   • Change agent behaviors through configuration
   • Swap processing strategies without code changes
   • Update workflow paths dynamically

Agents are autonomous components that perform specific tasks:

• Starter Agents: Handle input and validation
• Processor Agents: Transform and process data
• End Agents: Format and output results
• Error Handlers: Manage failures and recovery

Workflows define how agents interact:

• Stages: Sequential processing steps
• Transitions: Rules for moving between stages
• Error Paths: Alternative routes for handling failures

Everything is defined in JSON:

• Agent Definitions: Capabilities and settings
• Workflow Rules: Processing stages and paths
• System Settings: Global configurations

Here's a complete multi-agent system defined purely in configuration:

{
    "system_config": {
        "name": "Document Processing MAS",
        "version": "1.0.0"
    },
    "agents": {
        "document_reader": {
            "id": "reader_001",
            "type": "document_reader",
            "config": {
                "input_validation": {
                    "required_fields": ["text", "metadata"]
                }
            }
        },
        "text_processor": {
            "id": "processor_001",
            "type": "document_processor",
            "config": {
                "transformation_type": "uppercase",
                "max_retries": 3
            }
        },
        "document_writer": {
            "id": "writer_001",
            "type": "document_writer",
            "config": {
                "output_format": "json"
            }
        }
    },
    "workflow_definitions": {
        "document_processing": {
            "stages": [
                {
                    "name": "read",
                    "agent": "document_reader",
                    "next_stage": "process",
                    "error_stage": "error_handling"
                },
                {
                    "name": "process",
                    "agent": "text_processor",
                    "next_stage": "write",
                    "error_stage": "error_handling"
                },
                {
                    "name": "write",
                    "agent": "document_writer"
                }
            ]
        }
    }
}

Run your multi-agent system with just a few lines of code:

from mas.workflow import WorkflowManager
import json

# Load your configuration
with open("config.json", "r") as f:
    config = json.load(f)

# Create and run your multi-agent system
workflow_manager = WorkflowManager(config)
result = workflow_manager.start_workflow(
    "document_processing",
    {"text": "Process this document", "metadata": {"type": "article"}}
)

The repository includes several examples to help you get started:

Located in examples/document_processing/, this example demonstrates:

• Input validation
• Text transformation
• Error handling
• Multi-stage workflow

To run the example:

python examples/document_processing/test_document_processing.py

The example includes:

• config.json: Complete system configuration
• test_document_processing.py: Test cases and runner
• README.md: Detailed documentation

Located in examples/data_pipeline/, this example shows how to build a data processing pipeline with:

• Schema validation
• Data normalization
• Statistical aggregation
• Formatted output

To run the example:

python examples/data_pipeline/test_data_pipeline.py

Example features:

• Multi-stage data processing
• Numeric data transformation
• Group-based aggregation
• Error handling with retries
• Pretty-printed JSON output

Configuration example:

{
    "agents": {
        "data_validator": {
            "type": "data_validator",
            "config": {
                "input_validation": {
                    "required_fields": ["data", "schema_version"],
                    "schema": {
                        "type": "object",
                        "properties": {
                            "data": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "name": {"type": "string"},
                                        "value": {"type": "number"}
                                    }
                                }
                            }
                        }
                    }
                }
            }
        },
        "data_transformer": {
            "type": "data_transformer",
            "config": {
                "transformation_type": "normalize",
                "normalization": {
                    "method": "min_max",
                    "target_range": [0, 1]
                }
            }
        }
    }
}

• Configuration-First Design

  • Define entire system behavior through JSON
  • No agent implementation code needed
  • Easy to modify and experiment

• Pre-built Agent Types

  • Document Processing Agents
  • Data Pipeline Agents
  • Error Handler Agents
  • Custom Agent Support

• Declarative Workflows

  • Define complex workflows in JSON
  • Automatic stage transitions
  • Built-in error handling paths

• Built-in Transformations

  • Text processing
  • Data validation
  • Numerical operations
  • Statistical aggregations
  • Custom transformations

• Use descriptive agent IDs
• Include version information
• Document configuration schema
• Validate configurations

• Define error stages
• Set retry policies
• Log failures
• Implement recovery strategies

• Create comprehensive test cases
• Test edge cases
• Validate configurations
• Monitor performance

• Create custom agents
• Add transformation types
• Implement new validators
• Extend base classes

pip install distributedapps-mas

• Python 3.8+
• numpy>=1.24.0
• jsonschema>=4.17.3
• typing-extensions>=4.8.0
• python-json-logger>=2.0.7

from mas.agent import Agent
from typing import Dict

class CustomProcessor(Agent):
    def process_message(self, message: Dict) -> Dict:
        # Your custom logic here
        return processed_data

Add it to your configuration:

{
    "agents": {
        "custom_processor": {
            "type": "custom",
            "class": "path.to.CustomProcessor",
            "config": {
                "your_settings": "here"
            }
        }
    }
}

def on_stage_complete(stage_info):
    print(f"Stage {stage_info['name']} completed")

workflow_manager.start_workflow(
    "your_workflow",
    data,
    callbacks={
        "on_stage_complete": on_stage_complete,
        "on_error": handle_error
    }
)

# Update configuration at runtime
config["agents"]["processor"]["config"].update({
    "transformation_type": "new_type"
})

# Reload workflow
workflow_manager = WorkflowManager(config)

We welcome contributions! Whether it's:

• Adding new agent types
• Creating transformation plugins
• Improving documentation
• Reporting bugs

See our Contributing Guidelines for details.

Full documentation is available at https://github.com/kenhuangus/mas/wiki

• Documentation
• Discord Community
• Issue Tracker
• Email Support

This project is licensed under the MIT License - see the LICENSE file for details.

Ken Huang
CEO, Distributedapps.ai
ken@distributedapps.ai

Made with by Distributedapps.ai