Skip to content

Getting Started

Getting Started with PyRulesEngine

PyRulesEngine is an industry-grade package for abstracting rules and running rule evaluations dynamically and asynchronously. Furthermore, the repository functions natively as a complete Model Context Protocol (MCP) Server for LLM Agentic Workflows.

Publicly accessible modules, interfaces, and methods

There are several core classes that you will interact with. The primary interaction is via the RulesEngine instance, along with StorageManager to handle configuration states.

Rules

The rules in this system are purely evaluated string expressions based on Google's Common Expression Language (CEL). They are completely decoupled from unsafe runtime evaluation boundaries, using celpy natively wrapped in recursion protectors. CEL is non-Turing complete and sandboxed by design.

Rules Schema

The workflow rules define how we store the logical configuration. For py_rules_engine, the configuration is formatted identically to Microsoft's JDM (JSON Decision Model), but can also natively utilize YAML files (recommended).

An example yaml would be:

- WorkflowName: "Discount"
  Rules:
    - RuleName: "GiveDiscount10"
      SuccessEvent: "10"
      ErrorMessage: "One or more adjust rules failed."
      ErrorType: "Error"
      RuleExpressionType: "LambdaExpression"
      Expression: "input1.country == 'india' && input1.loyaltyFactor <= 2 && input1.totalPurchasesToDate >= 5000 && input2.totalOrders > 2"

RuleResultTree

This model is the immutable output of the PyRulesEngine. Once execution of the Rules Engine is completed through all the nodes, a list of this typed model is generated. It includes:

Rule

This is the executed rule context block.

IsSuccess

A boolean representing if the final AST constraints passed.

ChildResults

If the rule invokes sub-rules, this contains a nested hierarchy of RuleResultTree nodes.

The Engine Execute Wrapper

The core method is async and bound directly to RulesEngine: 

async def execute_all_rules_async(self, workflow_name: str, inputs: dict, version: str = "latest"):

  • workflow_name is the top level group node.
  • inputs is the dynamic dictionary passed into evaluating variables.
  • version defines what model snapshot to map. Defaults to "latest".

Initiating the Rules Engine

Instantiating requires injecting the desired Storage Provider.

from rules_engine import RulesEngine, StorageManager
from rules_engine.storage.file_system import FileStorageProvider

storage_manager = StorageManager()
storage_manager.register_provider(FileStorageProvider("./rules_configs"))

engine = RulesEngine(storage_manager)

Success/Failure Result Handling

Rules engine evaluations return structured responses.

results = await engine.execute_all_rules_async("Discount", {"input1": {"country": "india"}, ...})

total_discount = 0
for res in results:
    if res.is_success:
        total_discount += int(res.success_event)

How to use PyRulesEngine in Python

  1. Ensure Python 3.11+ is installed.
  2. Instantiate StorageManager with a FileStorageProvider.
  3. Pass the Manager into RulesEngine().
  4. Create dynamically generated dict inputs.
  5. await engine.execute_all_rules_async(workflow, inputs)
  6. Process the output sequence nodes.

Agentic System Integration via MCP

In addition to native async evaluation, py_rules_engine exposes a fully autonomous set of tools via an MCP Server. This allows AI assistants like Claude Desktop, Cursor, or your custom orchestration layer to natively converse, generate, manage, and execute complex business logic schemas continuously using natural language!

  1. Select your target backend: RULES_ENGINE__LLM_BACKEND (openai, anthropic, gemini)
  2. Pass your API Keys (e.g., ANTHROPIC_API_KEY).
  3. Deploy the application via docker-compose up or your preferred MCP client over stdio via mcp run rules_engine_mcp/server.py.
  4. The Agentic framework will now successfully map: Let your agents query, write, and execute logic automatically!