Validation API

from marsys.coordination.validation import ValidationProcessor

ValidationProcessor(
    topology_graph: TopologyGraph,
    response_format: str = "json"
)

async def process_response(
    raw_response: Any,
    agent: BaseAgent,
    branch: ExecutionBranch,
    exec_state: ExecutionState
) -> ValidationResult

processor = ValidationProcessor(topology_graph)

result = await processor.process_response(
    raw_response={"next_action": "invoke_agent", "action_input": "Analyzer"},
    agent=coordinator_agent,
    branch=current_branch,
    exec_state=execution_state
)

if result.is_valid:
    print(f"Action: {result.action_type}")
    print(f"Next agents: {result.next_agents}")

from marsys.coordination.validation import ValidationResult

if result.is_valid:
    if result.action_type == ActionType.INVOKE_AGENT:
        next_agent = result.next_agents[0]
        print(f"Invoking: {next_agent}")
    elif result.action_type == ActionType.PARALLEL_INVOKE:
        print(f"Parallel invoke: {result.next_agents}")
    elif result.action_type == ActionType.FINAL_RESPONSE:
        print(f"Final: {result.parsed_response['content']}")

from marsys.coordination.validation import ActionType

from marsys.coordination.routing import Router

Router(topology_graph: TopologyGraph)

async def route(
    validation_result: ValidationResult,
    current_branch: ExecutionBranch,
    routing_context: RoutingContext
) -> RoutingDecision

router = Router(topology_graph)

decision = await router.route(
    validation_result=validation_result,
    current_branch=current_branch,
    routing_context=RoutingContext(
        metadata={"retry_count": 0},
        error_info=None
    )
)

# Process routing decision
for step in decision.next_steps:
    if step.step_type == StepType.AGENT_INVOCATION:
        await invoke_agent(step.target)

from marsys.coordination.routing import RoutingDecision

decision = RoutingDecision(
    next_steps=[
        ExecutionStep(
            step_type=StepType.AGENT_INVOCATION,
            target="Analyzer",
            data={"request": "Analyze data"}
        )
    ],
    should_continue=True,
    branch_specs=[],
    metadata={"step_count": 5}
)

from marsys.coordination.routing import RoutingContext

from marsys.coordination.routing import ExecutionStep, StepType

class StepType(Enum):
    AGENT_INVOCATION = "agent_invocation"
    TOOL_EXECUTION = "tool_execution"
    PARALLEL_SPAWN = "parallel_spawn"
    WAIT_FOR_CONVERGENCE = "wait_for_convergence"
    FINAL_RESPONSE = "final_response"
    ERROR_RECOVERY = "error_recovery"

# Sequential invocation
{
    "thought": "I need to analyze this data",
    "next_action": "invoke_agent",
    "action_input": "DataAnalyzer"
}

# With request data
{
    "next_action": "invoke_agent",
    "action_input": "DataAnalyzer",
    "request": "Analyze sales data for Q4"
}

{
    "thought": "These can run in parallel",
    "next_action": "parallel_invoke",
    "agents": ["Worker1", "Worker2", "Worker3"],
    "agent_requests": {
        "Worker1": "Process segment A",
        "Worker2": "Process segment B",
        "Worker3": "Process segment C"
    }
}

{
    "next_action": "call_tool",
    "tool_calls": [
        {
            "id": "call_123",
            "type": "function",
            "function": {
                "name": "search",
                "arguments": "{\"query\": \"AI trends\"}"
            }
        }
    ]
}

# Text response
{
    "next_action": "final_response",
    "content": "Here is the analysis result..."
}

# Structured response
{
    "next_action": "final_response",
    "content": {
        "title": "Analysis Report",
        "sections": [...],
        "conclusion": "..."
    }
}

{
    "next_action": "error_recovery",
    "error_details": {
        "type": "api_quota_exceeded",
        "message": "OpenAI API quota exceeded",
        "provider": "openai"
    },
    "suggested_action": "retry"
}

from marsys.coordination.formats import (
    SystemPromptBuilder,
    BaseResponseFormat,
    JSONResponseFormat,
    AgentContext,
    CoordinationContext,
)

from marsys.coordination.formats import SystemPromptBuilder

SystemPromptBuilder(response_format: str = "json")

def build(
    agent_context: AgentContext,
    coordination_context: CoordinationContext,
    environmental: Optional[dict] = None
) -> str

builder = SystemPromptBuilder(response_format="json")

system_prompt = builder.build(
    agent_context=AgentContext(
        name="Coordinator",
        goal="Coordinate tasks",
        instruction="You coordinate worker agents..."
    ),
    coordination_context=CoordinationContext(
        next_agents=["Worker1", "Worker2"],
        can_return_final_response=True
    )
)

from marsys.coordination.formats import AgentContext

from marsys.coordination.formats import CoordinationContext

from marsys.coordination.formats import (
    register_format,
    get_format,
    list_formats,
    set_default_format,
    is_format_registered
)

# List available formats
formats = list_formats()  # ["json"]

# Get specific format
json_format = get_format("json")

# Register custom format
register_format("xml", XMLResponseFormat)

from marsys.coordination.formats import BaseResponseFormat

# Structured JSON Processor
class StructuredJSONProcessor(ResponseProcessor):
    """Handles JSON responses with next_action structure."""

    def can_process(self, response: Any) -> bool:
        return isinstance(response, dict) and "next_action" in response

    def priority(self) -> int:
        return 80  # Below error and tool processors

# Tool Call Processor
class ToolCallProcessor(ResponseProcessor):
    """Handles native tool call responses."""

    def can_process(self, response: Any) -> bool:
        return hasattr(response, 'tool_calls')

    def priority(self) -> int:
        return 90

# Error Message Processor
class ErrorMessageProcessor(ResponseProcessor):
    """Handles error Messages from agents."""

    def can_process(self, response: Any) -> bool:
        return isinstance(response, Message) and response.role == "error"

    def priority(self) -> int:
        return 100  # Highest priority

from marsys.coordination.formats import ResponseProcessor

class CustomFormatProcessor(ResponseProcessor):
    """Process custom response format."""

    def can_process(self, response: Any) -> bool:
        return isinstance(response, dict) and "custom_action" in response

    def process(self, response: Any) -> Optional[Dict[str, Any]]:
        return {
            "next_action": self._map_action(response["custom_action"]),
            "content": response.get("data")
        }

    def priority(self) -> int:
        return 75  # Between JSON and tool processors

# Register processor
validation_processor.register_processor(CustomFormatProcessor())

# 1. Receive agent response
response = await agent.run(prompt)

# 2. Process response
validation_result = await validation_processor.process_response(
    raw_response=response,
    agent=agent,
    branch=current_branch,
    exec_state=execution_state
)

# 3. Route based on validation
if validation_result.is_valid:
    routing_decision = await router.route(
        validation_result=validation_result,
        current_branch=current_branch,
        routing_context=context
    )

    # 4. Execute next steps
    for step in routing_decision.next_steps:
        await execute_step(step)
else:
    # Handle validation error
    logger.error(f"Validation failed: {validation_result.error_message}")
    if validation_result.retry_suggestion:
        # Apply steering for retry
        await apply_steering(validation_result.retry_suggestion)

if not result.is_valid:
    error_type = result.error_message

    if "not allowed" in error_type:
        # Permission denied - agent not in topology
        logger.error(f"Permission denied: {error_type}")

    elif "format" in error_type:
        # Invalid response format
        logger.error(f"Format error: {error_type}")

        # Use retry suggestion
        if result.retry_suggestion:
            steering = f"Please retry with: {result.retry_suggestion}"

    elif "missing" in error_type:
        # Missing required fields
        logger.error(f"Missing fields: {error_type}")

# Agent can trigger error recovery
response = {
    "next_action": "error_recovery",
    "error_details": {
        "type": "rate_limit",
        "message": "API rate limit exceeded",
        "retry_after": 60
    },
    "suggested_action": "wait_and_retry"
}

# Routes to User node for intervention