API Reference

This reference documents the Multi-Agent Framework's Python API for developers who want to extend or integrate with MAF.

Table of Contents

• Core Classes
• Agent Base Classes
• Event System
• Model Providers
• Utilities
• Extension Points
• Examples

Core Classes

AgentFactory

Factory class for creating and managing agents.

from multi_agent_framework.core.agent_factory import AgentFactory

class AgentFactory:
    """Creates and manages agent instances."""
    
    @staticmethod
    def create_agent(agent_type: str, config: Dict[str, Any]) -> BaseAgent:
        """
        Create an agent instance.
        
        Args:
            agent_type: Type of agent (e.g., 'frontend_agent', 'backend_agent')
            config: Agent configuration dictionary
            
        Returns:
            BaseAgent: Configured agent instance
            
        Raises:
            ValueError: If agent_type is not recognized
        """
        
    @staticmethod
    def get_available_agents() -> List[str]:
        """Get list of available agent types."""
        
    @staticmethod
    def register_agent(agent_type: str, agent_class: Type[BaseAgent]) -> None:
        """Register a custom agent type."""

ProjectConfig

Manages project configuration and detection.

from multi_agent_framework.core.project_config import ProjectConfig

class ProjectConfig:
    """Project configuration management."""
    
    def __init__(self, project_root: str):
        """Initialize with project root directory."""
        
    def detect_project_type(self) -> str:
        """Auto-detect project type from files."""
        
    def load_config(self, config_path: str = None) -> Dict[str, Any]:
        """Load configuration from file."""
        
    def save_config(self, config: Dict[str, Any], config_path: str = None) -> None:
        """Save configuration to file."""
        
    @property
    def technology_stack(self) -> Dict[str, List[str]]:
        """Get detected technology stack."""

EventBus

Central event management system.

from multi_agent_framework.core.event_bus import EventBus, Event, EventType

class EventBus:
    """Asynchronous event bus for agent communication."""
    
    async def publish(self, event: Event) -> None:
        """
        Publish an event to all subscribers.
        
        Args:
            event: Event instance to publish
        """
        
    async def subscribe(
        self, 
        event_types: List[EventType], 
        callback: Callable[Event], Awaitable[None](/micgo/maf-standalone/wiki/Event],-Awaitable[None)
    ) -> str:
        """
        Subscribe to specific event types.
        
        Args:
            event_types: List of event types to subscribe to
            callback: Async function to call when event occurs
            
        Returns:
            str: Subscription ID
        """
        
    async def unsubscribe(self, subscription_id: str) -> None:
        """Unsubscribe from events."""
        
    async def get_history(
        self, 
        event_type: EventType = None,
        limit: int = 100
    ) -> List[Event]:
        """Get event history with optional filtering."""

Agent Base Classes

BaseAgent

Abstract base class for all agents.

from multi_agent_framework.agents.base_agent import BaseAgent

class BaseAgent(ABC):
    """Base class for all agents."""
    
    def __init__(self, name: str, config: Dict[str, Any]):
        """
        Initialize agent.
        
        Args:
            name: Unique agent name
            config: Agent configuration
        """
        self.name = name
        self.config = config
        self.model_provider = self._init_model_provider()
        
    @abstractmethod
    async def process_task(self, task: Task) -> TaskResult:
        """
        Process a task and return result.
        
        Args:
            task: Task to process
            
        Returns:
            TaskResult: Processing result
        """
        
    async def initialize(self) -> None:
        """Initialize agent resources."""
        
    async def shutdown(self) -> None:
        """Clean up agent resources."""

EventDrivenBaseAgent

Base class for event-driven agents.

from multi_agent_framework.agents.event_driven_base_agent import EventDrivenBaseAgent

class EventDrivenBaseAgent(BaseAgent):
    """Base class for event-driven agents."""
    
    def __init__(self, name: str, config: Dict[str, Any], event_bus: EventBus):
        """Initialize with event bus."""
        super().__init__(name, config)
        self.event_bus = event_bus
        
    async def start(self) -> None:
        """Start listening for events."""
        
    async def stop(self) -> None:
        """Stop listening for events."""
        
    async def handle_event(self, event: Event) -> None:
        """
        Handle incoming event.
        
        Args:
            event: Event to handle
        """
        
    async def publish_result(self, task_id: str, result: Any) -> None:
        """Publish task result as event."""

Event System

Event Types

from multi_agent_framework.core.events import EventType

class EventType(Enum):
    """Available event types."""
    
    # Task lifecycle
    TASK_CREATED = "task_created"
    TASK_ASSIGNED = "task_assigned"
    TASK_STARTED = "task_started"
    TASK_COMPLETED = "task_completed"
    TASK_FAILED = "task_failed"
    
    # Code generation
    CODE_GENERATED = "code_generated"
    CODE_REVIEWED = "code_reviewed"
    CODE_INTEGRATED = "code_integrated"
    
    # System events
    AGENT_STARTED = "agent_started"
    AGENT_STOPPED = "agent_stopped"
    SYSTEM_ERROR = "system_error"

Event Class

from multi_agent_framework.core.events import Event

@dataclass
class Event:
    """Event data structure."""
    
    event_type: EventType
    source_agent: str
    target_agent: Optional[str] = None
    task_id: Optional[str] = None
    payload: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.utcnow)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary."""
        
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Event':
        """Create from dictionary."""

Task Class

from multi_agent_framework.core.task import Task, TaskStatus

@dataclass
class Task:
    """Task data structure."""
    
    id: str
    description: str
    task_type: str
    assigned_agent: Optional[str] = None
    status: TaskStatus = TaskStatus.PENDING
    priority: int = 1
    dependencies: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.utcnow)
    
    def can_start(self, completed_tasks: Set[str]) -> bool:
        """Check if task can start based on dependencies."""

Model Providers

Base Provider

from multi_agent_framework.llm.base_provider import BaseModelProvider

class BaseModelProvider(ABC):
    """Abstract base class for LLM providers."""
    
    @abstractmethod
    async def generate(
        self,
        prompt: str,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        **kwargs
    ) -> str:
        """
        Generate text from prompt.
        
        Args:
            prompt: Input prompt
            temperature: Sampling temperature
            max_tokens: Maximum tokens to generate
            **kwargs: Provider-specific parameters
            
        Returns:
            str: Generated text
        """
        
    @abstractmethod
    def validate_config(self) -> bool:
        """Validate provider configuration."""

Provider Factory

from multi_agent_framework.llm.provider_factory import ProviderFactory

class ProviderFactory:
    """Factory for creating model providers."""
    
    @staticmethod
    def create_provider(
        provider_type: str,
        api_key: str,
        model_name: str,
        **kwargs
    ) -> BaseModelProvider:
        """
        Create a model provider instance.
        
        Args:
            provider_type: 'gemini', 'anthropic', or 'openai'
            api_key: API key for the provider
            model_name: Model identifier
            **kwargs: Additional provider-specific config
            
        Returns:
            BaseModelProvider: Configured provider instance
        """

Utilities

FileIntegrator

Smart file integration utilities.

from multi_agent_framework.core.file_integrator import FileIntegrator

class FileIntegrator:
    """Integrates code from multiple agents."""
    
    def __init__(self, project_root: str):
        """Initialize with project root."""
        
    async def integrate_files(
        self,
        file_updates: List[FileUpdate],
        strategy: str = "smart"
    ) -> IntegrationResult:
        """
        Integrate multiple file updates.
        
        Args:
            file_updates: List of file updates to integrate
            strategy: Integration strategy ('smart', 'overwrite', 'merge')
            
        Returns:
            IntegrationResult: Result with conflicts if any
        """
        
    def detect_conflicts(
        self,
        existing_content: str,
        new_content: str
    ) -> List[Conflict]:
        """Detect conflicts between file versions."""

CrossAgentValidator

Validates compatibility between agent outputs.

from multi_agent_framework.core.cross_agent_validator import CrossAgentValidator

class CrossAgentValidator:
    """Validates cross-agent compatibility."""
    
    async def validate_integration(
        self,
        frontend_code: Dict[str, str],
        backend_code: Dict[str, str],
        database_schema: Dict[str, Any]
    ) -> ValidationResult:
        """
        Validate integration points between components.
        
        Returns:
            ValidationResult: Validation results with any issues
        """
        
    def check_api_compatibility(
        self,
        frontend_calls: List[APICall],
        backend_endpoints: List[Endpoint]
    ) -> List[CompatibilityIssue]:
        """Check API compatibility between frontend and backend."""

Extension Points

Custom Agent Creation

from multi_agent_framework.agents import EventDrivenBaseAgent
from multi_agent_framework.core import EventType, Event

class CustomAgent(EventDrivenBaseAgent):
    """Example custom agent implementation."""
    
    def __init__(self, name: str, config: Dict[str, Any], event_bus: EventBus):
        super().__init__(name, config, event_bus)
        # Custom initialization
        
    async def process_task(self, task: Task) -> TaskResult:
        """Process tasks specific to this agent."""
        try:
            # Custom task processing logic
            prompt = self._build_prompt(task)
            result = await self.model_provider.generate(prompt)
            
            # Publish completion event
            await self.publish_result(task.id, result)
            
            return TaskResult(
                task_id=task.id,
                success=True,
                output=result
            )
        except Exception as e:
            return TaskResult(
                task_id=task.id,
                success=False,
                error=str(e)
            )
    
    def _build_prompt(self, task: Task) -> str:
        """Build prompt for the task."""
        return f"Custom prompt for: {task.description}"

# Register the custom agent
AgentFactory.register_agent("custom_agent", CustomAgent)

Custom Event Types

# Define custom event types
class CustomEventType(Enum):
    CUSTOM_ACTION = "custom_action"
    CUSTOM_NOTIFICATION = "custom_notification"

# Use in your agent
async def publish_custom_event(self):
    event = Event(
        event_type=CustomEventType.CUSTOM_ACTION,
        source_agent=self.name,
        payload={"action": "custom_work", "data": {...}}
    )
    await self.event_bus.publish(event)

Custom Model Provider

from multi_agent_framework.llm import BaseModelProvider

class CustomModelProvider(BaseModelProvider):
    """Custom LLM provider implementation."""
    
    def __init__(self, api_key: str, model_name: str, **kwargs):
        self.api_key = api_key
        self.model_name = model_name
        # Initialize your client
        
    async def generate(
        self,
        prompt: str,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        **kwargs
    ) -> str:
        """Generate text using custom model."""
        # Your implementation
        response = await self.client.complete(
            prompt=prompt,
            temperature=temperature,
            max_tokens=max_tokens
        )
        return response.text
        
    def validate_config(self) -> bool:
        """Validate provider configuration."""
        return bool(self.api_key) and bool(self.model_name)

Examples

Creating a Custom Integration

import asyncio
from multi_agent_framework import AgentFactory, EventBus, ProjectConfig

async def main():
    # Initialize components
    project_config = ProjectConfig("/path/to/project")
    event_bus = EventBus()
    
    # Create agents
    orchestrator = AgentFactory.create_agent(
        "orchestrator",
        config=project_config.get_agent_config("orchestrator")
    )
    
    custom_agent = AgentFactory.create_agent(
        "custom_agent",
        config={"model_provider": "gemini", "model_name": "gemini-pro"}
    )
    
    # Start agents
    await orchestrator.start()
    await custom_agent.start()
    
    # Create a task
    from multi_agent_framework.core import Task
    task = Task(
        id="task_001",
        description="Custom processing task",
        task_type="custom",
        assigned_agent="custom_agent"
    )
    
    # Publish task event
    from multi_agent_framework.core import Event, EventType
    event = Event(
        event_type=EventType.TASK_CREATED,
        source_agent="api",
        target_agent="orchestrator",
        task_id=task.id,
        payload={"task": task.to_dict()}
    )
    
    await event_bus.publish(event)
    
    # Wait for completion
    await asyncio.sleep(10)
    
    # Shutdown
    await orchestrator.stop()
    await custom_agent.stop()

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

Subscribing to Events

from multi_agent_framework import EventBus, EventType

async def handle_task_completed(event: Event):
    """Handle task completion events."""
    print(f"Task {event.task_id} completed by {event.source_agent}")
    result = event.payload.get("result")
    # Process the result

# Subscribe to events
event_bus = EventBus()
subscription_id = await event_bus.subscribe(
    [EventType.TASK_COMPLETED],
    handle_task_completed
)

# Later, unsubscribe
await event_bus.unsubscribe(subscription_id)

Direct Agent Usage

from multi_agent_framework.agents.specialized import BackendAgent

# Create agent directly
agent = BackendAgent(
    name="backend",
    config={
        "model_provider": "gemini",
        "model_name": "gemini-2.0-flash-exp",
        "framework": "fastapi"
    }
)

# Process a task
task = Task(
    id="api_001",
    description="Create user registration endpoint",
    task_type="api_endpoint"
)

result = await agent.process_task(task)
print(f"Generated code: {result.output}")

Best Practices

1. Agent Development

• Always inherit from appropriate base class
• Implement proper error handling
• Use type hints for clarity
• Document your agent's capabilities

2. Event Handling

• Subscribe only to relevant events
• Handle events asynchronously
• Clean up subscriptions on shutdown
• Use event metadata for context

3. Task Processing

• Validate task inputs
• Report progress for long tasks
• Handle timeouts gracefully
• Include error details in failures

4. Model Integration

• Use appropriate temperature settings
• Implement retry logic
• Cache responses when possible
• Monitor token usage

Next Steps

• See Architecture for system design
• Read Agent Reference for agent details
• Check Examples for more code samples
• Join Discussions for questions