Moonchild 🌙

Moonchild is a declarative Python workflow framework with a built-in real-time UI.

About this repository

Welcome to the repository of Moonchild, developed by Ally Bros. Moonchild lets you define multi-step workflows in pure Python — with parallel branches, conditional logic, human-in-the-loop wait states, and cron scheduling — then serves a real-time dashboard so you can monitor and interact with every run. You can use this source code under the terms of the MIT License.

🌟 Features

• Declarative fluent builder: .when().step().and_().or_().if_().finish()
• Parallel branches: .and_() (all complete) / .or_() (first wins)
• Conditional branching: .if_(condition).then_(a).else_(b).end_if()
• Human-in-the-loop wait actions with auto-rendered UI forms
• Cron and manual triggers
• Real-time dashboard over WebSocket
• Zero-setup TinyDB persistence
• Publish once, run anywhere — just pip install moonchild

📀 Installation

pip install moonchild

Requires Python 3.11+.

🕹 Usage

1. Define your actions

from moonchild import Action, Result

class FetchData(Action):
    def execute(self, ctx):
        ctx["data"] = "hello world"
        return Result.next()

class ProcessData(Action):
    def execute(self, ctx):
        ctx["result"] = {"value": ctx["data"].upper()}
        return Result.next()

2. Build your workflow

from moonchild import workflow, ManualTrigger

workflow("my-pipeline") \
    .when(ManualTrigger()) \
    .step(FetchData(),   name="fetch") \
    .step(ProcessData(), name="process") \
    .finish()

3. Start

from moonchild import Moonchild

Moonchild.start()   # serves at http://localhost:8000

Full example (app.py):

from moonchild import Moonchild, workflow, Action, Result, ManualTrigger

class FetchData(Action):
    def execute(self, ctx):
        ctx["data"] = "hello world"
        return Result.next()

workflow("my-pipeline") \
    .when(ManualTrigger()) \
    .step(FetchData(), name="fetch") \
    .finish()

Moonchild.start()

python app.py

🔀 Parallel branches

from moonchild import workflow, branch, ManualTrigger

workflow("parallel-wf") \
    .when(ManualTrigger()) \
    .step(Ingest(), name="ingest") \
    .and_(                                      # wait for ALL branches
        branch("validate")
            .step(ValidateSchema(), name="validate-schema")
            .step(ValidateRules(),  name="validate-rules"),
        branch("enrich")
            .step(EnrichGeo(),  name="enrich-geo")
            .step(EnrichTags(), name="enrich-tags"),
        name="validate-and-enrich",
    ) \
    .or_(                                       # continue on FIRST branch done
        branch("email").step(SendEmail(), name="email"),
        branch("sms").step(SendSMS(),     name="sms"),
        name="notify",
    ) \
    .step(Finalize(), name="finalize") \
    .finish()

Branches run in parallel threads. The shared ctx dict is lock-protected.

🔁 Conditional branching

from moonchild import Action, Result

class IsWeekend(Action):
    def execute(self, ctx):
        import datetime
        return Result.branch(datetime.date.today().weekday() >= 5)

workflow("conditional-wf") \
    .when(ManualTrigger()) \
    .if_(IsWeekend(), name="check-weekend")
        .then_(WeekendPipeline(), name="weekend")
        .else_(WeekdayPipeline(), name="weekday")
    .end_if() \
    .step(Finalize(), name="finalize") \
    .finish()

⏸ Human-in-the-loop

from moonchild import WaitAction, WaitHint, Field, HintAction, Result

class ReviewAction(WaitAction):
    def execute(self, ctx):
        ctx["_wait_hint"] = WaitHint(
            title="Review required",
            body="Please review the item below.",
            fields=[Field(key="note", type="textarea", label="Note")],
            actions=[
                HintAction(key="approve", label="Approve", style="success"),
                HintAction(key="reject",  label="Reject",  style="danger"),
            ],
        ).to_dict()
        return Result.next()

    def resume(self, ctx, input):
        ctx["decision"] = input.get("action")
        if ctx["decision"] == "reject":
            return Result.goto("fetch")
        return Result.next()

The dashboard renders the form automatically — no frontend code needed.

For simple prompts use the built-in:

from moonchild import PromptWaitAction

.step(PromptWaitAction("What is your name?", key="name"), name="ask")

⏰ Triggers

from moonchild import ManualTrigger, CronTrigger

ManualTrigger()              # start from UI or API
CronTrigger("*/5 * * * *")  # every 5 minutes
CronTrigger("0 9 * * 1-5")  # 9am weekdays

⚙️ Configuration

Moonchild.configure(
    db_path="./data/moonchild.json",
    host="0.0.0.0",
    port=8000,
)
Moonchild.start()

🌐 REST API

📁 Project layout

your-project/
├── app.py              # entry point
├── workflows/
│   ├── billing.py
│   ├── onboarding.py
│   └── reports.py
└── requirements.txt    # moonchild

👾 Tech Stack

• Python 3.11+
• FastAPI + Uvicorn
• TinyDB
• WebSocket (real-time UI)
• Vanilla JS + SVG (zero frontend dependencies)

📄 License

GPL-3.0