Execution API Reference

from marsys.coordination.execution import BranchExecutor

BranchExecutor(
    validation_processor: ValidationProcessor,
    router: Router,
    step_executor: StepExecutor,
    context_manager: ContextManager,
    config: ExecutionConfig
)

async def execute_branch(
    branch: ExecutionBranch,
    initial_request: Any,
    context: Dict[str, Any],
    resume_with_results: Optional[Dict] = None
) -> BranchResult

executor = BranchExecutor(
    validation_processor=validator,
    router=router,
    step_executor=step_exec,
    context_manager=ctx_mgr,
    config=exec_config
)

result = await executor.execute_branch(
    branch=main_branch,
    initial_request="Analyze this data",
    context={"session_id": "123"}
)

from marsys.coordination.branches.types import ExecutionBranch, BranchType, BranchStatus

class BranchType(Enum):
    SIMPLE = "simple"                    # Sequential execution
    CONVERSATION = "conversation"        # Bidirectional dialogue
    NESTED = "nested"                    # Has child branches
    USER_INTERACTION = "user_interaction"  # Human-in-the-loop

class BranchStatus(Enum):
    PENDING = "pending"      # Not started
    RUNNING = "running"      # Currently executing
    WAITING = "waiting"      # Waiting for child branches
    COMPLETED = "completed"  # Successfully finished
    FAILED = "failed"        # Terminated with error

from marsys.coordination.branches.types import ExecutionBranch, BranchType

branch = ExecutionBranch(
    branch_id="main_001",
    branch_type=BranchType.SIMPLE,
    agent_sequence=["Coordinator", "Worker"],
    metadata={"priority": "high"}
)

from marsys.coordination.execution import StepExecutor

StepExecutor(
    config: Optional[ExecutionConfig] = None,
    tool_executor: Optional[ToolExecutor] = None,
    user_node_handler: Optional[UserNodeHandler] = None,
    event_bus: Optional[EventBus] = None,
    system_prompt_builder: Optional[SystemPromptBuilder] = None,
    max_retries: int = 5,
    retry_delay: float = 1.0
)

async def execute_step(
    agent: Union[BaseAgent, str],
    request: Any,
    memory: List[Dict[str, Any]],
    context: Dict[str, Any]
) -> StepResult

@dataclass
class StepResult:
    """Result of executing a single step within a branch."""
    # Core execution fields (set by StepExecutor)
    agent_name: str
    success: bool
    response: Any = None

    # Execution metadata (set by StepExecutor)
    step_id: Optional[str] = None
    memory_updates: List[Dict[str, Any]] = field(default_factory=list)
    tool_calls: List[Dict[str, Any]] = field(default_factory=list)
    tool_results: List[Dict[str, Any]] = field(default_factory=list)
    error: Optional[str] = None
    requires_retry: bool = False
    context_selection: Optional[Dict[str, Any]] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

    # Routing decision fields (set by BranchExecutor ONLY)
    action_type: Optional[str] = None
    parsed_response: Optional[Dict[str, Any]] = None
    next_agent: Optional[str] = None
    should_end_branch: bool = False
    waiting_for_children: bool = False
    child_branch_ids: List[str] = field(default_factory=list)
    convergence_target: Optional[str] = None

from marsys.coordination.formats import SystemPromptBuilder

step_executor = StepExecutor(
    config=exec_config,
    system_prompt_builder=SystemPromptBuilder(response_format="json")
)

result = await step_executor.execute_step(
    agent="Analyzer",
    request="Analyze sales data",
    memory=conversation_memory,
    context={"session": "123"}
)

from marsys.coordination.execution import DynamicBranchSpawner

DynamicBranchSpawner(
    branch_executor: BranchExecutor,
    context_manager: ContextManager,
    config: ExecutionConfig
)

async def handle_parallel_invocation(
    agents: List[str],
    requests: Dict[str, Any],
    parent_branch: ExecutionBranch,
    context: Dict[str, Any]
) -> List[asyncio.Task]

spawner = DynamicBranchSpawner(
    branch_executor=branch_exec,
    context_manager=ctx_mgr,
    config=exec_config
)

# Spawn parallel branches
tasks = await spawner.handle_parallel_invocation(
    agents=["Worker1", "Worker2", "Worker3"],
    requests={
        "Worker1": "Task 1",
        "Worker2": "Task 2",
        "Worker3": "Task 3"
    },
    parent_branch=main_branch,
    context={"parallel": True}
)

# Wait for completion
results = await asyncio.gather(*tasks)

from marsys.coordination.branches.types import BranchResult

if result.status == BranchStatus.COMPLETED:
    print(f"Success: {result.final_response}")
    print(f"Executed {result.steps_executed} steps")
    print(f"Time: {result.execution_time:.2f}s")
else:
    print(f"Failed: {result.error}")

from marsys.coordination.execution import ToolExecutor

async def execute_tool_calls(
    tool_calls: List[ToolCall],
    agent_name: str,
    context: Dict[str, Any]
) -> List[ToolResult]

tool_executor = ToolExecutor(
    agent_registry=AgentRegistry,
    config=exec_config
)

results = await tool_executor.execute_tool_calls(
    tool_calls=[
        ToolCall(name="search", arguments={"query": "AI news"}),
        ToolCall(name="summarize", arguments={"text": "..."})
    ],
    agent_name="Researcher",
    context={}
)

# Simple sequential branch
branch = ExecutionBranch(
    branch_type=BranchType.SIMPLE,
    agent_sequence=["Agent1", "Agent2", "Agent3"]
)

result = await executor.execute_branch(
    branch=branch,
    initial_request="Process this",
    context={}
)

# Bidirectional conversation
branch = ExecutionBranch(
    branch_type=BranchType.CONVERSATION,
    agent_sequence=["Agent1", "Agent2"],
    metadata={"max_turns": 5}
)

result = await executor.execute_branch(
    branch=branch,
    initial_request="Let's discuss",
    context={}
)

# Branch with child branches
parent_branch = ExecutionBranch(
    branch_type=BranchType.NESTED,
    agent_sequence=["Coordinator"]
)

# Coordinator spawns child branches dynamically
result = await executor.execute_branch(
    branch=parent_branch,
    initial_request="Coordinate tasks",
    context={"allow_parallel": True}
)

# Response that triggers parallel execution
response = {
    "next_action": "parallel_invoke",
    "agents": ["Worker1", "Worker2", "Worker3"],
    "agent_requests": {
        "Worker1": "Analyze segment A",
        "Worker2": "Analyze segment B",
        "Worker3": "Analyze segment C"
    }
}

# DynamicBranchSpawner handles this automatically

# Wait for all parallel branches
results = await spawner.wait_for_convergence(
    child_branch_ids=["branch_1", "branch_2", "branch_3"],
    timeout=300.0  # 5 minutes
)

# Resume parent with aggregated results
parent_result = await executor.execute_branch(
    branch=parent_branch,
    initial_request=None,
    context=context,
    resume_with_results=results
)