Deploy AI Agents in Production: Long-Running Python Agents

Most agents are written as a request and a reply: you call them, they answer, they're gone. The agents that do real work in production look different. They wake up on a schedule or when something happens, do one bounded job, remember what matters for next time, and stop themselves before they do too much. This guide builds one of those in Python with the Promptise Agent Runtime: a support-ticket triage agent that runs every few minutes, keeps state between runs, writes a journal you can read back, and pauses itself when it goes over its budget. Every snippet ran against Promptise Foundry 1.3.1 with gpt-5-mini, and the output you see is what those runs printed, including the parts that didn't go to plan.

Run the agent as a long-lived process, not a function call. Give it a trigger that wakes it (a cron schedule, a webhook, a file landing in a folder), tools that reach your systems, limits on what one run may do, and a record of what it did. With Promptise Foundry, that process is an AgentProcess:

import asyncio
import sys

from promptise.config import StdioServerSpec
from promptise.runtime import AgentProcess, ProcessConfig, TriggerConfig


async def main() -> None:
    process = AgentProcess(
        name="triage",
        config=ProcessConfig(
            model="openai:gpt-5-mini",
            instructions="Triage new support tickets, then post one short summary.",
            servers={"tickets": StdioServerSpec(command=sys.executable, args=["tickets_server.py"])},
            triggers=[TriggerConfig(type="cron", cron_expression="*/15 * * * *")],
        ),
    )
    await process.start()  # builds the agent, connects its tools, starts the trigger
    print(process.status()["state"], "with", process.status()["trigger_count"], "trigger")
    try:
        await asyncio.Event().wait()  # from here on, the agent wakes every 15 minutes
    finally:
        await process.stop()
        print(process.status()["state"])


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

running with 1 trigger
stopped

That's a working AI agent cron job. start() builds the agent, connects to its MCP tools and starts the trigger. From then on the cron trigger wakes it every 15 minutes, with no ainvoke() call from you, until you stop it. Everything else in this guide is what you add before you trust it to run unattended.

An AgentProcess wraps an ordinary Promptise agent. Triggers put events on a queue. A worker takes each event, adds what the agent should know (its context state, recent conversation, remaining budget) and runs the agent once. The budget is checked before every tool call, and after the run the runtime decides whether the process keeps going.

Each process moves through a small state machine, and every transition is validated:

A suspended process keeps its triggers. Events that fire while it's paused wait in the queue until someone calls resume().

• Python 3.10 or newer.

• Promptise Foundry: pip install "promptise>=1.3.1". Cron, webhook and file-watch triggers work out of the box; their dependencies ship with the base install.

• An API key for a model provider. This guide uses OpenAI with openai:gpt-5-mini; set OPENAI_API_KEY in your environment or a .env file.

The agent's job: every time it wakes, find new support tickets, give each a priority, and post one summary to the team channel, naming the on-call engineer for anything urgent.

The agent reaches your systems through MCP tools. Here a small MCP server stands in for a ticketing system and a team chat: a JSON file of tickets, and a log file as the channel. If you haven't built one before, How to Connect MCP Servers to Your AI Agent in Python covers the basics.

import json
from datetime import datetime
from pathlib import Path
from typing import Literal

from promptise.mcp.server import MCPServer, ToolError

server = MCPServer("tickets")

# A stand-in for your ticketing system: a JSON file the agent reads and updates.
TICKETS = Path(__file__).with_name("tickets.json")
CHANNEL = Path(__file__).with_name("support-channel.log")


def load() -> dict:
    return json.loads(TICKETS.read_text()) if TICKETS.exists() else {}


@server.tool(read_only_hint=True)
async def list_untriaged() -> list[dict]:
    """List new support tickets that don't have a priority yet."""
    return [{"ticket_id": tid, **t} for tid, t in load().items() if t.get("priority") is None]


@server.tool()
async def set_priority(ticket_id: str, priority: Literal["P1", "P2", "P3"], reason: str) -> dict:
    """Give a ticket a priority. P1 = outage or data loss, P2 = a customer is blocked, P3 = everything else.

    Args:
        reason: One short sentence explaining the priority.
    """
    tickets = load()
    if ticket_id not in tickets:
        raise ToolError(f"There is no ticket {ticket_id}.")
    tickets[ticket_id].update(priority=priority, reason=reason)
    TICKETS.write_text(json.dumps(tickets, indent=2))
    return {"ticket_id": ticket_id, "priority": priority}


@server.tool()
async def post_summary(text: str) -> str:
    """Post a short triage summary to the support team's channel."""
    stamp = datetime.now().strftime("%H:%M:%S")
    with CHANNEL.open("a") as f:
        f.write(f"[{stamp}] {text}\n")
    return "posted"


if __name__ == "__main__":
    server.run()

Notice where the durable state lives: in the ticket system. A ticket with a priority is done, so the next run won't touch it again. That's the most reliable memory an agent can have, and it's yours, not the runtime's.

ProcessConfig holds everything about the process: model, instructions, tools, triggers, starting state and limits. The file also writes a journal, which Step 6 explains.

import asyncio
import sys

from promptise import CallbackSink, EventNotifier
from promptise.config import StdioServerSpec
from promptise.runtime import (
    AgentProcess,
    BudgetConfig,
    ContextConfig,
    FileJournal,
    JournalEntry,
    ProcessConfig,
    ToolCostAnnotation,
    TriggerConfig,
)

INSTRUCTIONS = """\
You triage the support queue. Each time you wake up:
1. Call list_untriaged.
2. Give every ticket a priority with set_priority.
3. If you set any priorities, post ONE short summary with post_summary.
   For every P1, name the on-call engineer: use the exact on_call value
   from [Context State]. Never invent a name.
If there are no new tickets, don't post anything and reply "Nothing to triage."
"""

CONFIG = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions=INSTRUCTIONS,
    servers={
        "tickets": StdioServerSpec(command=sys.executable, args=["tickets_server.py"]),
    },
    # Every minute, for the demo. In production: "*/15 * * * *" or "0 8 * * 1-5".
    triggers=[TriggerConfig(type="cron", cron_expression="* * * * *")],
    # State the agent sees at the start of every run.
    context={"initial_state": {"on_call": "Sam"}},
    # The autonomy budget: what one run, and one day, may do.
    budget=BudgetConfig(
        enabled=True,
        max_tool_calls_per_run=6,
        max_runs_per_day=200,
        tool_costs={"post_summary": ToolCostAnnotation(cost_weight=1.0, irreversible=True)},
        max_irreversible_per_run=1,
        on_exceeded="pause",
    ),
)

journal = FileJournal(".promptise/journal")


def build_process(name: str = "triage", initial_state: dict | None = None) -> AgentProcess:
    config = CONFIG
    if initial_state is not None:
        config = CONFIG.model_copy(update={"context": ContextConfig(initial_state=initial_state)})

    async def record(event) -> None:
        """Write every runtime event to the journal, plus a checkpoint after each run."""
        await journal.append(JournalEntry(process_id=name, entry_type=event.event_type, data=event.data))
        if event.event_type == "invocation.complete":
            await journal.checkpoint(
                name,
                {
                    "context_state": process.context.state_snapshot(),
                    "lifecycle_state": process.state.value,
                    "budget_remaining": process.status().get("budget"),
                },
            )

    process = AgentProcess(
        name=name,
        process_id=name,
        config=config,
        event_notifier=EventNotifier(sinks=[CallbackSink(record)]),
    )
    return process


async def main() -> None:
    process = build_process()
    await process.start()
    print(f"{process.name} is {process.state.value}. Press Ctrl+C to stop.")
    try:
        while True:
            await asyncio.sleep(3600)
    finally:
        await process.stop()
        print(f"{process.name} is {process.state.value}.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

The parts that matter:

• The trigger. A standard five-field cron expression. This guide uses every minute so you can watch several cycles; a real triage agent might run every 15 minutes, or at 8:00 on weekdays.

• The context. initial_state is the process's key-value state. The runtime shows it to the agent at the start of every run, and your code can change it while the process runs.

• The budget. At most six tool calls per run, 200 runs a day, and one irreversible action per run. post_summary is marked irreversible because a message, once posted, can't be unposted. Cost weights are abstract units you choose, not money.

python triage_agent.py runs the process until you press Ctrl+C:

triage is running. Press Ctrl+C to stop.
triage is stopped.

To see the whole lifecycle in a few minutes, this script starts the process, adds tickets between runs and prints what happens. A small helper, seed_tickets.py, adds the tickets: two at first, one between runs, then a burst of five.

"""Drive the triage process through three cron cycles and print what happens."""

import asyncio
import subprocess
import sys
import time
from pathlib import Path

from triage_agent import build_process

CHANNEL = Path("support-channel.log")


def seed(batch: str) -> None:
    out = subprocess.run([sys.executable, "seed_tickets.py", batch], capture_output=True, text=True)
    print(f"{time.strftime('%H:%M:%S')}  queue: {out.stdout.strip()}")


async def wait_for_run(process, n: int) -> None:
    while process.status()["invocation_count"] < n:
        await asyncio.sleep(1)
    await asyncio.sleep(1)  # let post-run bookkeeping finish
    s = process.status()
    print(
        f"{time.strftime('%H:%M:%S')}  run {n} done: state={s['state']} "
        f"conversation_messages={s['conversation_messages']} "
        f"tool_calls_left_this_run={s['budget']['tool_calls_run']}"
    )


async def main() -> None:
    CHANNEL.unlink(missing_ok=True)
    seed("1")
    process = build_process()
    await process.start()
    print(f"{time.strftime('%H:%M:%S')}  {process.name} is {process.state.value}")

    await wait_for_run(process, 1)

    # Shift change: tell the running agent who is on call now.
    process.context.put("on_call", "Dana", source="operator")
    seed("2")
    await wait_for_run(process, 2)

    # A burst of tickets: more work than one run's budget allows.
    seed("burst")
    await wait_for_run(process, 3)

    # Wait for the next cron tick while paused.
    await asyncio.sleep(65)
    s = process.status()
    print(f"{time.strftime('%H:%M:%S')}  after next tick: state={s['state']} queue_size={s['queue_size']}")

    print("\nsupport-channel.log:")
    print(CHANNEL.read_text())
    print("on_call history:", [(e.value, e.source) for e in process.context.state_history("on_call")])

    await process.stop()
    print(f"{time.strftime('%H:%M:%S')}  {process.name} is {process.state.value}")


asyncio.run(main())

11:56:54  queue: 2 new ticket(s) in the queue
11:56:56  triage is running
11:57:09  run 1 done: state=running conversation_messages=2 tool_calls_left_this_run=2
11:57:09  queue: 1 new ticket(s) in the queue
11:58:11  run 2 done: state=running conversation_messages=4 tool_calls_left_this_run=3
11:58:11  queue: 5 new ticket(s) in the queue
Error in BudgetGuard.on_tool_start callback: BudgetExceededError("Budget limit max_tool_calls_per_run reached (limit 6); tool call 'post_summary' was not executed. Stop calling tools and give your final answer.")
Budget violation: max_tool_calls_per_run (limit=6, current=7) — action=pause
11:59:22  run 3 done: state=suspended conversation_messages=6 tool_calls_left_this_run=0
12:00:27  after next tick: state=suspended queue_size=1

support-channel.log:
[11:57:06] T-201 set to P1 (site-wide checkout 500s) — on-call: Sam. T-202 set to P2 (customer blocked from downloading invoice PDF).
[11:58:08] T-203 → P1: Nobody in the EU region can log in — on-call: Dana.

on_call history: [('Sam', 'system'), ('Dana', 'operator')]
12:00:28  triage is stopped

Three runs, each at the top of a minute, with no call from the script. Starting the process took about two seconds here; start() builds the agent and connects its tools once, and every run reuses them. The next three steps read this output from top to bottom.

Between runs 1 and 2, the on-call engineer changed. The script told the running process with one line:

process.context.put("on_call", "Dana", source="operator")

Run 1 named Sam for its P1. Run 2 named Dana. Nothing restarted. The context keeps an audit trail too: state_history("on_call") shows who set each value, the initial system value and the operator change.

The process carries two kinds of memory between runs:

• Context state, the key-value store you just used. The runtime adds it to the start of every run as a [Context State] message.

• Recent conversation, a rolling buffer of each run's trigger message and the agent's final reply. That's why conversation_messages grows by two per run. The default keeps the last 100 messages; change it with conversation_max_messages in ContextConfig.

Here's what the model received in run 3, on its first call and on the compacted call after the five priorities were set, captured with a LangChain callback:

  MODEL SEES: system | [Context State] {'on_call': 'Dana'}
  MODEL SEES: system | You triage the support queue. Each time you wake up: 1. Call list_untriaged. 2. Give every tick…
  MODEL SEES: system | [Budget Remaining] {'tool_calls_run': 6, 'llm_turns_run': None, 'cost_run': None, 'irreversible…
  MODEL SEES: human | [Trigger: cron] trigger_id=cron-f3e5fea9 event_id=6752f9fd-7c62-4261-b518-beb60e92651d at=2026-…
  MODEL SEES: ai | Triage complete. - T-201 → P1: Checkout is returning 500 errors for all customers since 09:10 —…
  MODEL SEES: human | [Trigger: cron] trigger_id=cron-f3e5fea9 event_id=e2830608-7e10-45a3-8321-229668c61364 at=2026-…
  MODEL SEES: ai | Triage complete. - T-203 → P1: Nobody in the EU region can log in — on-call: Dana.
  MODEL SEES: human | [Trigger: cron] trigger_id=cron-f3e5fea9 event_id=c08a0d74-d5e3-4cc1-930c-aef47ef7e699 at=2026-…
  ---
…
  ---
  MODEL SEES: system | [Context State] {'on_call': 'Dana'}
  MODEL SEES: system | You triage the support queue. Each time you wake up: 1. Call list_untriaged. 2. Give every tick…
  MODEL SEES: system | [Budget Remaining] {'tool_calls_run': 6, 'llm_turns_run': None, 'cost_run': None, 'irreversible…
  MODEL SEES: system | Earlier in this conversation (shortened). Answer the user's current request, which is the next …
  MODEL SEES: human | [Trigger: cron] trigger_id=cron-f3e5fea9 event_id=c08a0d74-d5e3-4cc1-930c-aef47ef7e699 at=2026-…
  MODEL SEES: ai | 
  MODEL SEES: tool | {"ticket_id": "T-300", "priority": "P2"}
  MODEL SEES: tool | {"ticket_id": "T-301", "priority": "P3"}
  MODEL SEES: tool | {"ticket_id": "T-302", "priority": "P2"}
  MODEL SEES: tool | {"ticket_id": "T-303", "priority": "P3"}
  MODEL SEES: tool | {"ticket_id": "T-304", "priority": "P2"}
  MODEL SEES: system | Facts already gathered by your earlier tool calls (do not call a tool again for these unless yo…

The first call also shows the recent conversation: the two earlier runs' trigger messages and replies. Since 1.3.0, a trigger's payload arrives inside an <untrusted-trigger-payload-…> block with a random tag, after a notice that its content is data from outside the system, not instructions.

Run 3 got five tickets at once. Listing them, prioritizing all five and posting the summary takes seven tool calls; the budget allows six. Here's what happened, from the output above:

• The run listed the tickets and set all five priorities, in one turn of five parallel calls. That's six tool calls.

• The seventh, post_summary, was refused before it ran. LangChain logged the refusal as Error in BudgetGuard.on_tool_start callback: BudgetExceededError(...), and the agent got the same message as the tool's result. No summary was posted: the channel log has no line from run 3.

• After the run, the runtime logged Budget violation: max_tool_calls_per_run (limit=6, current=7) — action=pause and suspended the process. The next cron tick still fired. Its event waited in the queue (queue_size=1) instead of running.

That's the most important thing to know about the autonomy budget: per-run limits are checked before every tool call, so a call that would cross one never runs. The run itself isn't cut off. The refusal tells the agent to stop calling tools and give its final answer, so it finishes with what it has done so far: here, five priorities and no summary. The daily limit, max_runs_per_day, is checked before a run starts, so the run over that limit never happens.

So set per-run limits above what a normal run needs: here, six calls covered the burst's priorities but not its summary. Use the daily limits to cap the total. Three more things worth knowing:

• on_exceeded can be "pause", "stop" or "escalate". "escalate" posts the violation to a webhook or an event bus, then pauses.

• Nothing resumes a paused process automatically, not even the daily reset. A person, or your own code, calls process.resume(), and the queued events run then.

• With inject_remaining=True (the default), the agent sees its remaining budget at the start of each run as a [Budget Remaining] message.

When something goes wrong at 3 a.m., you want a record of what the agent did. Promptise has a journal for exactly this: FileJournal writes append-only JSONL, one file per process, and promptise runtime logs prints it.

The process emits events as it works: process.started, invocation.start, invocation.complete, budget.warning, budget.exceeded and more. Pass an EventNotifier with a CallbackSink, and you can journal every one of them. This is the record function from Step 2:

    async def record(event) -> None:
        """Write every runtime event to the journal, plus a checkpoint after each run."""
        await journal.append(JournalEntry(process_id=name, entry_type=event.event_type, data=event.data))
        if event.event_type == "invocation.complete":
            await journal.checkpoint(
                name,
                {
                    "context_state": process.context.state_snapshot(),
                    "lifecycle_state": process.state.value,
                    "budget_remaining": process.status().get("budget"),
                },
            )

After each run it also stores a checkpoint: a snapshot of the context state, the lifecycle state and the remaining budget. Read the journal back from the command line:

promptise runtime logs triage --lines 40

                                Journal: triage                                 
┏━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Time                    ┃ Type                 ┃ Data                        ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ 2026-10-11T09:56:56.143 │ process.started      │ {"process_name": "triage",  │
│                         │                      │ "process_id": "triage"}     │
│ 2026-10-11T09:57:00.010 │ invocation.start     │ {"model":                   │
│                         │                      │ "openai:gpt-5-mini"}        │
│ 2026-10-11T09:57:07.672 │ invocation.complete  │ {"duration_ms": 7669.7}     │
│ 2026-10-11T09:57:07.673 │ checkpoint           │ {"state": {"context_state": │
│                         │                      │ {"on_call": "Sam"},         │
│                         │                      │ "lifecycle_state":          │
…
│ 2026-10-11T09:59:21.845 │ invocation.complete  │ {"duration_ms": 21361.6}    │
│ 2026-10-11T09:59:21.862 │ checkpoint           │ {"state": {"context_state": │
│                         │                      │ {"on_call": "Dana"},        │
│                         │                      │ "lifecycle_state":          │
│                         │                      │ "suspende...                │
│ 2026-10-11T09:59:21.862 │ budget.exceeded      │ {"process_name": "triage",  │
│                         │                      │ "limit_type":               │
│                         │                      │ "max_tool_calls_per_run",   │
│                         │                      │ "current":...               │
│ 2026-10-11T09:59:21.862 │ budget.warning       │ {"process_name": "triage",  │
│                         │                      │ "limit_type":               │
│                         │                      │ "max_tool_calls_per_run",   │
│                         │                      │ "current":...               │
│ 2026-10-11T10:00:27.981 │ process.stopped      │ {"process_name": "triage",  │
│                         │                      │ "process_id": "triage"}     │
└─────────────────────────┴──────────────────────┴─────────────────────────────┘

The full entries are in .promptise/journal/triage.jsonl, for example this line from run 3:

{"entry_id": "…", "process_id": "triage", "timestamp": "2026-10-11T09:59:21.862…", "entry_type": "budget.exceeded", "data": {"process_name": "triage", "limit_type": "max_tool_calls_per_run", "current": 7, "limit": 6}}

promptise runtime logs looks up the journal by the name you give it, which is why build_process uses the process name as its process_id too.

The process keeps its context in memory, so a restart starts from the initial state again. The checkpoint fixes that. ReplayEngine reads the journal and returns the last known state, and you start the new process with it:

import asyncio

from promptise.runtime import ReplayEngine

from triage_agent import build_process, journal


async def main() -> None:
    recovered = await ReplayEngine(journal).recover("triage")
    print("recovered:", recovered)

    process = build_process(initial_state=recovered["context_state"])
    await process.start()
    print("state after restart:", process.context.state_snapshot())
    await process.stop()


asyncio.run(main())

recovered: {'context_state': {'on_call': 'Dana'}, 'lifecycle_state': 'suspended', 'last_entry_type': 'process.stopped', 'entries_replayed': 3}
state after restart: {'on_call': 'Dana'}

The new process knows Dana is on call, although the config still says Sam. It also learned the old process was suspended when it stopped: here, paused by its budget. Starting it again is a decision for a person, so check lifecycle_state before you do.

A schedule is right for sweeps. When something else already knows a ticket arrived, let it wake the agent. A webhook trigger starts a small HTTP server inside the process; this keeps a slow cron sweep as a safety net:

import asyncio

from promptise.runtime import AgentProcess, TriggerConfig

from triage_agent import CONFIG

config = CONFIG.model_copy(
    update={
        "triggers": [
            TriggerConfig(type="cron", cron_expression="*/15 * * * *"),
            TriggerConfig(type="webhook", webhook_path="/tickets", webhook_port=8180),
        ]
    }
)


async def main() -> None:
    process = AgentProcess(name="triage", config=config)
    await process.start()
    print("listening on http://127.0.0.1:8180/tickets")
    while process.status()["invocation_count"] < 1:
        await asyncio.sleep(1)
    print("runs:", process.status()["invocation_count"])
    await process.stop()


asyncio.run(main())

Post an event the way your help desk would:

curl -s -X POST http://127.0.0.1:8180/tickets -H 'Content-Type: application/json' -d '{"event": "ticket.created", "ticket_id": "T-401"}'

{"status": "accepted", "event_id": "e4920f02-171f-48bf-b613-b0ff8695ba9a"}

The webhook answers 202 at once and the run happens in the background. The JSON body becomes the trigger payload the agent sees. The process printed:

WebhookTrigger on port 8180 has no HMAC secret — any HTTP client can trigger this webhook. Set hmac_secret for production use.
listening on http://127.0.0.1:8180/tickets
runs: 1

And the channel got:

[12:03:02] Triage update: 1 P1 — T-401: data export deleting rows (potential data loss). On-call: Sam.

Take that warning seriously. A webhook created from TriggerConfig listens on 127.0.0.1 unless you set webhook_host. Since 1.3.0, TriggerConfig also takes the signature settings: hmac_secret, signature_scheme (generic, github or stripe) and allowed_sources for an IP allowlist. Add a secret to the webhook trigger:

TriggerConfig(
    type="webhook",
    webhook_path="/tickets",
    webhook_port=8180,
    hmac_secret=os.environ["TICKETS_WEBHOOK_SECRET"],
)

In a test, an unsigned POST got 401 with {"status": "error", "message": "Missing or invalid signature"}, and a POST with a valid X-Webhook-Signature: sha256=<hex> header got 202. In a .agent manifest, write hmac_secret: ${TICKETS_WEBHOOK_SECRET} so the secret stays out of the file.

The other built-in triggers work the same way. file_watch wakes the agent when files matching watch_patterns appear or change in watch_path. event and message listen to an event bus or message broker, which you pass to an AgentRuntime to share between its processes. Triggers lists them all, and how to register your own.

You don't have to write Python to run a process. A .agent manifest describes the same thing in YAML, and the promptise runtime command runs it:

version: "1.0"
name: triage
model: openai:gpt-5-mini
instructions: |
  You triage the support queue. Each time you wake up:
  1. Call list_untriaged.
  2. Give every ticket a priority with set_priority.
  3. If you set any priorities, post ONE short summary with post_summary.
     For every P1, name the on-call engineer: use the exact on_call value
     from [Context State]. Never invent a name.
  If there are no new tickets, don't post anything and reply "Nothing to triage."

servers:
  tickets:
    type: stdio
    command: python
    args: ["tickets_server.py"]

triggers:
  - type: cron
    cron_expression: "* * * * *"   # every minute for the demo

world:
  on_call: Sam

budget:
  enabled: true
  max_tool_calls_per_run: 6
  max_runs_per_day: 200
  max_irreversible_per_run: 1
  on_exceeded: pause
  tool_costs:
    post_summary:
      cost_weight: 1.0
      irreversible: true

world is the manifest's name for the initial context state. Check the file before you run it:

promptise runtime validate triage.agent

Validating triage.agent...
✓ Schema validation passed
✓ No warnings
                           Manifest Summary                            
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Field        ┃ Value                                                ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ Name         │ triage                                               │
│ Model        │ openai:gpt-5-mini                                    │
│ Version      │ 1.0                                                  │
│ Instructions │ You triage the support queue. Each time you wake up: │
│              │ 1. Call...                                           │
│ Servers      │ 1                                                    │
│ Triggers     │ 1                                                    │
└──────────────┴──────────────────────────────────────────────────────┘
✓ Validation complete!

Then start it. It runs in the foreground until you press Ctrl+C:

promptise runtime start triage.agent

Loaded process: triage
                        Agent Processes                         
┏━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━┳━━━━━━━━━━┓
┃ Name   ┃ State   ┃ PID      ┃ Invocations ┃ Queue ┃ Uptime   ┃
┡━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━╇━━━━━━━━━━┩
│ triage │ RUNNING │ 4691fb53 │           0 │     0 │ 0h 0m 0s │
└────────┴─────────┴──────────┴─────────────┴───────┴──────────┘

Press Ctrl+C to stop all processes.

Shutting down...
All processes stopped.

In that run the agent woke at the next minute, triaged two tickets and posted its summary before Ctrl+C. Pass a directory instead of a file to start every .agent file in it, and promptise runtime init --template cron writes a starter manifest. The Agent Manifests and Runtime CLI docs have every field and option.

The budget limits how much an agent does. Three more opt-in subsystems watch what it does. All are off by default.

• Behavioral health (HealthConfig) spots bad patterns without calling a model: the same tool with the same arguments again and again, repeating loops, empty responses and high error rates. on_anomaly can log, pause or escalate.

• Mission (MissionConfig) gives a process an objective and success_criteria. Every eval_every runs, a model judges progress; the process can stop itself when the mission is achieved, or escalate when confidence drops below confidence_threshold.

• Secret scoping (SecretScopeConfig) gives each process its own secrets, resolved from ${ENV_VAR} references when it starts, encrypted in memory, with optional expiry, and revoked on stop(). A missing variable stops the start with KeyError: "Environment variable 'NOT_SET_ANYWHERE' not found (required by secret config)". The agent itself can only read them through the get_secret tool in open execution mode.

Tune health checks before you let them pause anything. Since 1.3.0, empty-response detection looks at the agent's final reply only, and stuck and loop detection look within one run, so an agent with nothing to do isn't flagged. With the default thresholds and on_anomaly="pause", the triage agent ran on an empty queue every five seconds:

run 1: state=running anomalies=0
run 2: state=running anomalies=0
run 3: state=running anomalies=0
run 4: state=running anomalies=0

Still, start with on_anomaly="log", read what it flags for a few days, then tune empty_threshold and stuck_threshold to fit your agent before switching to "pause". See Behavioral Health, Mission Model and Secret Scoping.

Know where every piece of state lives. In Promptise 1.3:

With that in mind:

• Plan for restarts at two levels. Since 1.3.0, restart_policy="on_failure" restarts a process that reached max_consecutive_failures, with backoff, up to max_restarts times. In a test with a model name that doesn't exist, every run failed: after three failures the process went to failed, restarted itself and was running again within ten seconds, twice, with restart_count going up each time. Still run the runtime itself under systemd, Docker or Kubernetes with a restart policy, restore context from your checkpoint on start, and alert on failed.

• Run one copy of each process. Each process has its own cron trigger and nothing coordinates replicas, so two copies of the same process both fire at every tick and do the work twice. Scale by giving different processes different jobs, and make tools that change data safe to repeat.

• Distributed means remote control, not failover. RuntimeTransport puts an HTTP API on a node's runtime (health, status, start, stop, inject an event). Since 1.3.0 it refuses to bind a public address without an auth_token, and RuntimeCoordinator(auth_token=...) sends the token to its nodes. RuntimeCoordinator registers nodes, checks their health and aggregates status. There's no shared state, no leader election and no automatic move of processes off a failed node.

• Keep runs small. Short runs stay inside the budget and are cheaper to retry.

• Lock down triggers. Give every webhook an hmac_secret, and put a reverse proxy in front if anything outside the machine calls it.

• Watch the journal. Alert on budget.exceeded, health.anomaly and invocation.error events; the CallbackSink from Step 6 is a good place to send them on.

How do I run an AI agent as a cron job?

Give an AgentProcess a trigger like TriggerConfig(type="cron", cron_expression="*/15 * * * *") and start it. The process stays up and wakes the agent on schedule, so the agent and its tool connections are built once instead of on every run. A plain system cron job that runs a script also works; it just rebuilds everything each time and keeps no state in between.

How does a long-running AI agent remember things between runs?

Three ways, in order of reliability. Your own systems, through tools: a triaged ticket stays triaged. The process context, a key-value store the agent sees at the start of every run, which you can checkpoint to a journal and restore. And a rolling buffer of recent trigger messages and replies. For anything that must survive a restart, use the first two.

How do I stop an autonomous AI agent from running up costs?

Enable BudgetConfig with per-run and daily limits on tool calls, LLM turns, cost weights and irreversible actions, and choose pause, stop or escalate for when a limit is crossed. Since 1.3.0, per-run limits are checked before every tool call, so a call that would cross one never runs, and max_runs_per_day is checked before a run starts. Track real token spend with your model provider as well.

Can I run AI agents on several machines?

You can run a runtime on each machine and manage them over HTTP with RuntimeTransport and RuntimeCoordinator. In Promptise 1.3 that gives you remote start, stop, status and health checks, with a bearer token, not shared state or automatic failover, so assign each process to one machine.

Do I need Kubernetes to deploy an AI agent?

No. One promptise runtime start in the foreground under any supervisor that restarts it on failure is enough to begin with: systemd, Docker with a restart policy, or a Kubernetes Deployment with one replica.

• Agent Runtime: the overview of processes, triggers, state and governance.

• Agent Processes: every ProcessConfig field and lifecycle method.

• Cron triggers and Triggers: schedules, webhooks, file watches and custom trigger types.

• Journal and Context & State: what to record and what the agent sees.

• Runtime Manager: several processes in one AgentRuntime.

• Build a Production MCP Server in Python: make the tools this agent calls safe for unattended use.

• OpenAPI to MCP: Turn Any REST API into an MCP Server: give the agent tools for an API you already have.