MARSYSFramework

Navigation

Getting StartedConceptsAPI ReferenceGuides
GitHub

Rules API

Complete API reference for the rules engine system that controls multi-agent execution flow through flexible constraints and policies.

Overview

The Rules API provides a powerful system for enforcing execution constraints, implementing business logic, and managing control flow in multi-agent workflows.

RulesEngine

Central engine for rule evaluation and enforcement.

Import

from marsys.coordination.rules import RulesEngine

Constructor

RulesEngine(
    rules: Optional[List[Rule]] = None,
    enable_conflict_resolution: bool = True,
    enable_caching: bool = True
)

Key Methods

add_rule

def add_rule(rule: Rule) -> None

Add a rule to the engine.

remove_rule

def remove_rule(rule_name: str) -> bool

Remove a rule by name. Returns True if found.

check_rules

async def check_rules(
    context: RuleContext,
    rule_type: Optional[RuleType] = None
) -> RuleResult

Check all applicable rules for the given context.

Example

engine = RulesEngine()


# Add rules
engine.add_rule(TimeoutRule(max_duration_seconds=300))
engine.add_rule(MaxAgentsRule(max_agents=10))


# Check rules
context = RuleContext(
    rule_type=RuleType.PRE_EXECUTION,
    session_id="session_123",
    elapsed_time=150.0
)


result = await engine.check_rules(context)
if not result.passed:
    print(f"Rule violation: {result.reason}")

Rule (Abstract Base)

Abstract base class for all rules.

Import

from marsys.coordination.rules import Rule, RuleType, RulePriority

Constructor

Rule(
    name: str,
    rule_type: RuleType,
    priority: RulePriority = RulePriority.NORMAL,
    enabled: bool = True
)

Abstract Methods

MethodReturnsDescription
check(context)RuleResultEvaluate rule against context
description()strGet human-readable description

RuleType

Types of rules in the system.

ValueDescriptionWhen Applied
PRE_EXECUTIONBefore branch/step executionValidation phase
POST_EXECUTIONAfter branch/step executionCleanup phase
SPAWN_CONTROLControl branch spawningBefore parallel spawn
RESOURCE_LIMITResource constraintsContinuous monitoring
FLOW_CONTROLExecution flow rulesRouting decisions
VALIDATIONData validation rulesInput/output validation

RulePriority

Rule execution priority levels.

ValuePriorityUse Case
CRITICAL100Security, safety rules
HIGH75Resource limits, timeouts
NORMAL50Standard business logic
LOW25Logging, metrics

RuleContext

Context passed to rules for evaluation.

Attributes

AttributeTypeDescription
rule_typeRuleTypeType of rule being evaluated
session_idstrCurrent session identifier
branchExecutionBranchCurrent branch (if applicable)
total_stepsintTotal steps executed
elapsed_timefloatElapsed time in seconds
active_agentsintNumber of active agents
memory_usage_mbfloatCurrent memory usage

RuleResult

Result of rule evaluation.

Attributes

AttributeTypeDescription
rule_namestrName of evaluated rule
passedboolWhether rule passed
actionstrAction to take (allow, block, modify, terminate)
reasonstrExplanation of result
modificationsDictSuggested modifications

Built-in Rules

TimeoutRule

from marsys.coordination.rules import TimeoutRule


rule = TimeoutRule(
    name="timeout_rule",
    max_duration_seconds=300.0,  # 5 minutes
    priority=RulePriority.HIGH
)

MaxAgentsRule

from marsys.coordination.rules import MaxAgentsRule


rule = MaxAgentsRule(
    name="max_agents_rule",
    max_agents=10,
    priority=RulePriority.HIGH
)

MaxStepsRule

from marsys.coordination.rules import MaxStepsRule


rule = MaxStepsRule(
    name="max_steps_rule",
    max_steps=100,
    priority=RulePriority.NORMAL
)

MemoryLimitRule

from marsys.coordination.rules import MemoryLimitRule


rule = MemoryLimitRule(
    name="memory_limit_rule",
    max_memory_mb=1024,  # 1GB
    priority=RulePriority.HIGH
)

ConditionalRule

from marsys.coordination.rules import ConditionalRule


rule = ConditionalRule(
    name="conditional_rule",
    condition=lambda ctx: ctx.metadata.get("premium_user", False),
    action_if_true="allow",
    action_if_false="block",
    priority=RulePriority.NORMAL
)

Custom Rules

from marsys.coordination.rules import Rule, RuleResult, RuleContext


class BusinessHoursRule(Rule):
    """Only allow execution during business hours."""


    def __init__(self):
        super().__init__(
            name="business_hours",
            rule_type=RuleType.PRE_EXECUTION,
            priority=RulePriority.NORMAL
        )


    async def check(self, context: RuleContext) -> RuleResult:
        from datetime import datetime


        hour = datetime.now().hour
        if 9 <= hour < 17:  # 9 AM to 5 PM
            return RuleResult(
                rule_name=self.name,
                passed=True,
                action="allow"
            )
        else:
            return RuleResult(
                rule_name=self.name,
                passed=False,
                action="block",
                reason="Outside business hours",
                suggestions={"retry_after": "9:00 AM"}
            )


    def description(self) -> str:
        return "Business hours enforcement (9 AM - 5 PM)"

Rule Patterns

Resource Management Pattern

# Create resource management rules
rules = [
    MemoryLimitRule(max_memory_mb=2048),
    MaxAgentsRule(max_agents=20),
    MaxStepsRule(max_steps=200)
]


engine = RulesEngine(rules=rules)

Time-based Pattern

# Create time-based rules
rules = [
    TimeoutRule(max_duration_seconds=600),
    AgentTimeoutRule(timeouts={"slow_agent": 120}),
    BusinessHoursRule()
]

Dynamic Rule Management

# Add rules at runtime
if user.is_premium:
    engine.add_rule(PremiumFeaturesRule())


# Disable rules temporarily
engine.get_rule("strict_timeout").enabled = False


# Remove rules
engine.remove_rule("development_only_rule")

Best Practices

Do

  • Use appropriate rule priorities
  • Provide clear reason messages in results
  • Cache expensive rule evaluations
  • Use rule metadata for debugging
  • Combine related rules into composite rules

Don't

  • Create rules with side effects in check()
  • Use blocking I/O in rule evaluation
  • Ignore rule priorities
  • Hard-code values in rules
  • Create circular rule dependencies

Pro Tip

Rules are evaluated in priority order. Use CRITICAL priority for security and safety rules that must always execute first.

Performance

Keep rule evaluation fast. Expensive operations should be cached or computed asynchronously.