Initial commit: Motia III Backend Setup

- iii-config.yaml mit Production-Settings (CORS, all interfaces)
- Ticketing-System Steps (create, triage, escalate, notify, SLA monitoring)
- Python dependencies via uv
- Systemd services für Motia Engine und iii Console
- README mit Deployment-Info

steps/__init__.py

@@ -0,0 +1 @@
+"""Motia iii Example Steps."""

steps/create_ticket_step.py

@@ -0,0 +1,61 @@
+"""Create Ticket Step - accepts a new support ticket via API and enqueues it for triage."""
+
+import random
+import string
+from datetime import datetime, timezone
+from typing import Any
+
+from motia import ApiRequest, ApiResponse, FlowContext, http
+
+config = {
+    "name": "CreateTicket",
+    "description": "Accepts a new support ticket via API and enqueues it for triage",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        http("POST", "/tickets"),
+    ],
+    "enqueues": ["ticket::created"],
+}
+
+
+async def handler(request: ApiRequest[dict[str, Any]], ctx: FlowContext[Any]) -> ApiResponse[Any]:
+    body = request.body or {}
+    title = body.get("title")
+    description = body.get("description")
+    priority = body.get("priority", "medium")
+    customer_email = body.get("customerEmail")
+
+    if not title or not description:
+        return ApiResponse(status=400, body={"error": "Title and description are required"})
+
+    random_suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=5))
+    ticket_id = f"TKT-{int(datetime.now(timezone.utc).timestamp() * 1000)}-{random_suffix}"
+
+    ticket = {
+        "id": ticket_id,
+        "title": title,
+        "description": description,
+        "priority": priority,
+        "customerEmail": customer_email,
+        "status": "open",
+        "createdAt": datetime.now(timezone.utc).isoformat(),
+    }
+
+    await ctx.state.set("tickets", ticket_id, ticket)
+    ctx.logger.info("Ticket created", {"ticketId": ticket_id, "priority": priority})
+
+    await ctx.enqueue({
+        "topic": "ticket::created",
+        "data": {
+            "ticketId": ticket_id,
+            "title": title,
+            "priority": priority,
+            "customerEmail": customer_email,
+        },
+    })
+
+    return ApiResponse(status=200, body={
+        "ticketId": ticket_id,
+        "status": "open",
+        "message": "Ticket created and queued for triage",
+    })

steps/escalate_ticket_step.py

@@ -0,0 +1,90 @@
+"""Escalate Ticket Step - multi-trigger: escalates tickets from SLA breach or manual request.
+
+Uses ctx.match() to route logic per trigger type.
+"""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from motia import ApiRequest, ApiResponse, FlowContext, http, queue
+
+config = {
+    "name": "EscalateTicket",
+    "description": "Multi-trigger: escalates tickets from SLA breach or manual request",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        queue("ticket::sla-breached"),
+        http("POST", "/tickets/escalate"),
+    ],
+    "enqueues": [],
+}
+
+
+async def _escalate_ticket(
+    ticket_id: str,
+    updates: dict[str, Any],
+    ctx: FlowContext[Any],
+) -> dict[str, Any] | None:
+    """Fetches a ticket and applies escalation fields to state. Returns pre-update ticket or None."""
+    existing = await ctx.state.get("tickets", ticket_id)
+    if not existing:
+        return None
+    await ctx.state.set("tickets", ticket_id, {
+        **existing,
+        "escalatedTo": "engineering-lead",
+        "escalatedAt": datetime.now(timezone.utc).isoformat(),
+        **updates,
+    })
+    return existing
+
+
+async def handler(input_data: Any, ctx: FlowContext[Any]) -> Any:
+    async def _queue_handler(breach: Any) -> None:
+        ticket_id = breach.get("ticketId")
+        age_minutes = breach.get("ageMinutes", 0)
+        priority = breach.get("priority", "medium")
+
+        ctx.logger.info("Escalating ticket", {"ticketId": ticket_id, "triggerType": "queue"})
+        ctx.logger.warn("Auto-escalation from SLA breach", {
+            "ticketId": ticket_id,
+            "ageMinutes": age_minutes,
+            "priority": priority,
+        })
+
+        escalated = await _escalate_ticket(
+            ticket_id,
+            {"escalationReason": f"SLA breach: {age_minutes} minutes without resolution", "escalationMethod": "auto"},
+            ctx,
+        )
+
+        if not escalated:
+            ctx.logger.error("Ticket not found during SLA escalation", {"ticketId": ticket_id, "ageMinutes": age_minutes})
+
+    async def _http_handler(request: ApiRequest[Any]) -> ApiResponse[Any]:
+        body = request.body or {}
+        ticket_id = body.get("ticketId")
+        reason = body.get("reason", "")
+
+        ctx.logger.info("Escalating ticket", {"ticketId": ticket_id, "triggerType": "http"})
+
+        existing = await _escalate_ticket(
+            ticket_id,
+            {"escalationReason": reason, "escalationMethod": "manual"},
+            ctx,
+        )
+
+        if not existing:
+            return ApiResponse(status=404, body={"error": f"Ticket {ticket_id} not found"})
+
+        ctx.logger.info("Manual escalation via API", {"ticketId": ticket_id, "reason": reason})
+
+        return ApiResponse(status=200, body={
+            "ticketId": ticket_id,
+            "escalatedTo": "engineering-lead",
+            "message": "Ticket escalated successfully",
+        })
+
+    return await ctx.match({
+        "queue": _queue_handler,
+        "http": _http_handler,
+    })

steps/list_tickets_step.py

@@ -0,0 +1,24 @@
+"""List Tickets Step - returns all tickets from state."""
+
+from typing import Any
+
+from motia import ApiRequest, ApiResponse, FlowContext, http
+
+config = {
+    "name": "ListTickets",
+    "description": "Returns all tickets from state",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        http("GET", "/tickets"),
+    ],
+    "enqueues": [],
+}
+
+
+async def handler(request: ApiRequest[Any], ctx: FlowContext[Any]) -> ApiResponse[Any]:
+    _ = request
+    tickets = await ctx.state.list("tickets")
+
+    ctx.logger.info("Listing tickets", {"count": len(tickets)})
+
+    return ApiResponse(status=200, body={"tickets": tickets, "count": len(tickets)})

steps/notify_customer_step.py

@@ -0,0 +1,37 @@
+"""Notify Customer Step - sends a notification when a ticket has been triaged."""
+
+import re
+from typing import Any
+
+from motia import FlowContext, queue
+
+config = {
+    "name": "NotifyCustomer",
+    "description": "Sends a notification when a ticket has been triaged",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        queue("ticket::triaged"),
+    ],
+    "enqueues": [],
+}
+
+
+async def handler(input_data: Any, ctx: FlowContext[Any]) -> None:
+    ticket_id = input_data.get("ticketId")
+    assignee = input_data.get("assignee")
+    priority = input_data.get("priority")
+    title = input_data.get("title")
+
+    ctx.logger.info("Sending customer notification", {"ticketId": ticket_id, "assignee": assignee})
+
+    ticket = await ctx.state.get("tickets", ticket_id)
+    customer_email = ticket.get("customerEmail", "") if ticket else ""
+    redacted_email = re.sub(r"(?<=.{2}).(?=.*@)", "*", customer_email) if customer_email else "unknown"
+
+    ctx.logger.info("Notification sent", {
+        "ticketId": ticket_id,
+        "assignee": assignee,
+        "priority": priority,
+        "title": title,
+        "email": redacted_email,
+    })

steps/sla_monitor_step.py

@@ -0,0 +1,67 @@
+"""SLA Monitor Step - cron job that checks for SLA breaches on open tickets."""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from motia import FlowContext, cron
+
+SLA_THRESHOLDS_MS = {
+    "critical": 15 * 60 * 1000,       # 15 minutes
+    "high": 60 * 60 * 1000,           # 1 hour
+    "medium": 4 * 60 * 60 * 1000,     # 4 hours
+    "low": 24 * 60 * 60 * 1000,       # 24 hours
+}
+
+config = {
+    "name": "SlaMonitor",
+    "description": "Cron job that checks for SLA breaches on open tickets",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        cron("0/30 * * * * *"),
+    ],
+    "enqueues": ["ticket::sla-breached"],
+}
+
+
+async def handler(input_data: None, ctx: FlowContext[Any]) -> None:
+    _ = input_data
+    ctx.logger.info("Running SLA compliance check")
+
+    tickets = await ctx.state.list("tickets")
+    now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
+    breaches = 0
+
+    for ticket in tickets:
+        if ticket.get("status") != "open" or not ticket.get("createdAt"):
+            continue
+
+        try:
+            created_dt = datetime.fromisoformat(ticket["createdAt"])
+            created_ms = int(created_dt.timestamp() * 1000)
+        except (ValueError, TypeError):
+            continue
+
+        age_ms = now_ms - created_ms
+        threshold = SLA_THRESHOLDS_MS.get(ticket.get("priority", "medium"), SLA_THRESHOLDS_MS["medium"])
+
+        if age_ms > threshold:
+            breaches += 1
+            age_minutes = round(age_ms / 60_000)
+
+            ctx.logger.warn("SLA breach detected!", {
+                "ticketId": ticket["id"],
+                "priority": ticket.get("priority"),
+                "ageMinutes": age_minutes,
+            })
+
+            await ctx.enqueue({
+                "topic": "ticket::sla-breached",
+                "data": {
+                    "ticketId": ticket["id"],
+                    "priority": ticket.get("priority", "medium"),
+                    "title": ticket.get("title", ""),
+                    "ageMinutes": age_minutes,
+                },
+            })
+
+    ctx.logger.info("SLA check complete", {"totalTickets": len(tickets), "breaches": breaches})

steps/triage_ticket_step.py

@@ -0,0 +1,100 @@
+"""Triage Ticket Step - multi-trigger: auto-triage from queue, manual triage via API, sweep via cron.
+
+Demonstrates a single step responding to three trigger types using ctx.match().
+"""
+
+from datetime import datetime, timezone
+from typing import Any
+
+from motia import ApiRequest, ApiResponse, FlowContext, cron, http, queue
+
+config = {
+    "name": "TriageTicket",
+    "description": "Multi-trigger: auto-triage from queue, manual triage via API, sweep via cron",
+    "flows": ["support-ticket-flow"],
+    "triggers": [
+        queue("ticket::created"),
+        http("POST", "/tickets/triage"),
+        cron("0 */5 * * * * *"),
+    ],
+    "enqueues": ["ticket::triaged"],
+}
+
+
+async def _triage_ticket(
+    ticket_id: str,
+    existing: dict[str, Any] | None,
+    state_updates: dict[str, Any],
+    enqueue_data: dict[str, Any],
+    ctx: FlowContext[Any],
+) -> None:
+    """Updates ticket state with triage fields and emits the triaged event."""
+    if not existing:
+        return
+    updated = {**existing, "triagedAt": datetime.now(timezone.utc).isoformat(), **state_updates}
+    await ctx.state.set("tickets", ticket_id, updated)
+    await ctx.enqueue({"topic": "ticket::triaged", "data": {"ticketId": ticket_id, **enqueue_data}})
+
+
+async def handler(input_data: Any, ctx: FlowContext[Any]) -> Any:
+    async def _queue_handler(data: Any) -> None:
+        ticket_id = data.get("ticketId")
+        title = data.get("title", "")
+        priority = data.get("priority", "medium")
+
+        ctx.logger.info("Auto-triaging ticket from queue", {"ticketId": ticket_id, "priority": priority})
+
+        assignee = "senior-support" if priority in ("critical", "high") else "support-pool"
+        existing = await ctx.state.get("tickets", ticket_id)
+
+        await _triage_ticket(
+            ticket_id, existing,
+            {"assignee": assignee, "triageMethod": "auto"},
+            {"assignee": assignee, "priority": priority, "title": title},
+            ctx,
+        )
+        ctx.logger.info("Ticket auto-triaged", {"ticketId": ticket_id, "assignee": assignee})
+
+    async def _http_handler(request: ApiRequest[Any]) -> ApiResponse[Any]:
+        body = request.body or {}
+        ticket_id = body.get("ticketId")
+        assignee = body.get("assignee")
+        priority = body.get("priority", "medium")
+
+        existing = await ctx.state.get("tickets", ticket_id)
+        if not existing:
+            return ApiResponse(status=404, body={"error": f"Ticket {ticket_id} not found"})
+
+        ctx.logger.info("Manual triage via API", {"ticketId": ticket_id, "assignee": assignee})
+
+        await _triage_ticket(
+            ticket_id, existing,
+            {"assignee": assignee, "priority": priority, "triageMethod": "manual"},
+            {"assignee": assignee, "priority": priority, "title": existing.get("title", "")},
+            ctx,
+        )
+        return ApiResponse(status=200, body={"ticketId": ticket_id, "assignee": assignee, "status": "triaged"})
+
+    async def _cron_handler() -> None:
+        ctx.logger.info("Running untriaged ticket sweep.")
+        tickets = await ctx.state.list("tickets")
+        swept = 0
+
+        for ticket in tickets:
+            if not ticket.get("assignee") and ticket.get("status") == "open":
+                ctx.logger.warn("Found untriaged ticket during sweep", {"ticketId": ticket["id"]})
+                await _triage_ticket(
+                    ticket["id"], ticket,
+                    {"assignee": "support-pool", "triageMethod": "auto-sweep"},
+                    {"assignee": "support-pool", "priority": ticket.get("priority", "medium"), "title": ticket.get("title", "unknown")},
+                    ctx,
+                )
+                swept += 1
+
+        ctx.logger.info("Sweep complete", {"sweptCount": swept})
+
+    return await ctx.match({
+        "queue": _queue_handler,
+        "http": _http_handler,
+        "cron": _cron_handler,
+    })