Run Multiple AI Agents: One Runtime, Many Processes

Most teams that run one autonomous agent soon run several: a triage agent on a webhook, a digest agent on a morning schedule, a reporter every hour. Running each as its own script means separate startup code, separate logs and separate shutdown handling. This guide shows how to run multiple AI agents together in one Promptise Foundry AgentRuntime: add and remove processes, load them from manifests, share one event notifier, keep a failure in one agent from touching the others, and shut everything down cleanly on SIGTERM.

Register each agent as a named process in one runtime, start them all, and let the runtime stop them all on exit. Each process keeps its own triggers, queue, workers, model client, tools and failure counters, so one agent failing doesn't stop the rest. The runtime adds shared pieces: one place to look up status, one notifier for events, and one shutdown path. In Promptise:

async with AgentRuntime(event_notifier=notifier) as runtime:
    await runtime.load_directory("agents")           # one .agent file per agent
    await runtime.add_process("reporter", ProcessConfig(...))
    await runtime.start_all()                        # a failed start is logged, not raised
    print(runtime.status())
    await stop_signal.wait()
# leaving the block calls stop_all()

Key takeaways:

• Processes are isolated in state, not in resources. Each has its own counters and lifecycle, but all share one Python program and one event loop.

• `start_all()` never raises for a single process. Check status() afterwards.

• Restart a `FAILED` process with `stop_process()` and then `start_process()` in 1.4.3, not restart_process().

• Handle SIGTERM yourself. The runtime cleans up on exit, but your program decides when that is.

• `AgentRuntime(config)` doesn't register processes. AgentRuntime.from_config(config) does.

An agent runtime does for AI agents what a process supervisor does for programs. It keeps a registry of named workers, starts and stops them, reports their state, and contains failures. The difference is that an agent "process" here is not an operating system process. It is a set of asyncio tasks inside one Python program: trigger listeners, a queue, workers that call the model, and a heartbeat.

AgentRuntime  (one Python program, one asyncio event loop)
│
├── shared: event_notifier ──▶ sinks (callback, log, webhook)
├── shared: event_bus, broker  (only if you pass them)
│
├── process "triage"     triggers ─▶ queue ─▶ workers ─▶ agent + MCP clients
│                        lifecycle, failure counters, restart policy, journal
├── process "digest"     triggers ─▶ queue ─▶ workers ─▶ agent + MCP clients
│                        lifecycle, failure counters, restart policy, journal
└── process "reporter"   triggers ─▶ queue ─▶ workers ─▶ agent + MCP clients
                         lifecycle, failure counters, restart policy, journal

What each process owns:

• Its configuration and agent. Model, instructions, tools, triggers, budget and health settings come from its own ProcessConfig. Two processes can use different models and different MCP servers.

• Its queue and workers. Events for one agent never wait behind another agent's events.

• Its lifecycle and failure handling. max_consecutive_failures, the restart policy and the state machine apply per process. One agent in FAILED is just one entry in the status table.

What they share:

• The event notifier, if you pass one to the runtime. Every process emits into it, and the runtime drains it once all processes have stopped.

• The event bus and message broker, if you pass them. These let one agent's output trigger another agent. Event-based chaining gets its own guide; here it is enough to know the runtime hands the same objects to every process.

• The Python program itself: CPU, memory, the event loop, and the program's lifetime.

One runtime suits you when:

• The agents belong to one product or team and are deployed together.

• You want one status view, one alert channel and one shutdown path.

• Most of the work is waiting on model and tool calls, which asyncio handles well.

Split them into separate programs or containers when:

• One agent needs a different Python environment, secrets boundary or network access.

• One agent does heavy CPU work in-process, which would slow every other agent.

• Agents need to be deployed, scaled or rolled back independently.

• A crash of one must not take down the others. Isolation inside a runtime covers failed runs, not a crashed program.

• Python 3.10 or later and Promptise Foundry 1.4.3 or later.

• No API key. The examples use the fake OpenAI-compatible model server from the auto-restart guide, flaky_llm.py, which can make one model fail while others work.

• About 20 minutes.

pip install "promptise>=1.4.3"

This guide builds on deploy AI agents in production, which walks through a single long-running agent.

Agents that change rarely are easiest to keep in .agent files, one per agent, in version control. A manifest has the same fields as ProcessConfig, plus a config: block for any field without its own section:

version: "1.0"
name: triage
model: openai:gpt-5-mini
instructions: |
  Triage new support tickets.
triggers:
  - type: cron
    cron_expression: "* * * * * */2"   # every 2 seconds (demo only)

version: "1.0"
name: digest
model: openai:gpt-5-mini
instructions: |
  Write a short digest of yesterday's tickets.
triggers:
  - type: cron
    cron_expression: "* * * * * */3"   # every 3 seconds (demo only)
config:
  restart_policy: on_failure
  max_restarts: 5

The six-field cron expressions fire every few seconds so the demo finishes quickly. Real agents use five fields, such as 0 7 * * *. AI agent YAML config files covers the format; promptise runtime validate agents/triage.agent checks a file before you deploy it.

The demo loads both manifests, adds two more processes in code, and starts everything. reporter uses a model that the fake server reports as down. archiver points at an MCP server command that doesn't exist, so it can't even start.

"""Run several agent processes in one AgentRuntime; one fails, the others keep working."""
import asyncio
import os
import time

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8765/v1"  # fake model server
os.environ.setdefault("OPENAI_API_KEY", "sk-fake")

from flaky_llm import FlakyLLM
from promptise.events import CallbackSink, EventNotifier
from promptise.runtime import AgentRuntime, ProcessConfig, RuntimeBaseError, TriggerConfig

T0 = time.monotonic()


def log(message: str) -> None:
    print(f"{time.monotonic() - T0:5.1f}s  {message}")


async def on_event(event) -> None:
    log(f"event {event.event_type:17} from {event.agent_id}")


def table(runtime: AgentRuntime) -> None:
    for name, s in runtime.status()["processes"].items():
        print(f"         {name:9} {s['state']:8} invocations={s['invocation_count']} "
              f"failures={s['consecutive_failures']} restarts={s['restart_count']}")


async def main() -> None:
    llm = FlakyLLM()
    llm.down_models = {"gpt-5-nano"}  # this model's provider is down
    await llm.start()

    notifier = EventNotifier(sinks=[CallbackSink(on_event, events=[
        "process.started", "process.failed", "process.stopped",
    ])])
    async with AgentRuntime(event_notifier=notifier) as runtime:
        names = await runtime.load_directory("agents")
        log(f"loaded from manifests: {names}")
        await runtime.add_process("reporter", ProcessConfig(
            model="openai:gpt-5-nano",
            instructions="Send the hourly report.",
            triggers=[TriggerConfig(type="cron", cron_expression="* * * * * *")],
            max_consecutive_failures=2,
        ))
        await runtime.add_process("archiver", ProcessConfig(
            model="openai:gpt-5-mini",
            servers={"files": {"command": "no-such-mcp-server"}},  # cannot start
        ))
        try:
            await runtime.add_process("triage", ProcessConfig())
        except RuntimeBaseError as exc:
            log(f"duplicate name: {exc}")

        await runtime.start_all()  # logs the archiver's failure, starts the rest
        await asyncio.sleep(4.5)
        log("status after 4.5 s:")
        table(runtime)

        llm.down_models.clear()  # the provider is back
        await runtime.stop_process("reporter")   # FAILED: stop() cleans up first
        await runtime.start_process("reporter")
        await runtime.restart_process("digest")  # RUNNING: stop, then start
        await runtime.remove_process("archiver")
        await asyncio.sleep(2.2)
        log("status after restarting reporter and digest, removing archiver:")
        table(runtime)
        log("leaving the async with block")
    log("stop_all() done")
    await llm.stop()


asyncio.run(main())

The calls that build the runtime:

• `load_directory("agents")` loads every *.agent file in sorted order and returns the names. load_manifest(path, name_override=...) loads one file. Both register processes; neither starts them.

• `add_process(name, config)` registers a process from code and returns the AgentProcess. Names must be unique: a second triage raises RuntimeBaseError.

• `start_all()` starts every process in CREATED, STOPPED or FAILED. If one fails to start, the error is logged and the others start anyway.

• `async with AgentRuntime(...)` calls stop_all() when the block ends, also when it ends with an exception.

The first part of the output:

  0.0s  loaded from manifests: ['digest', 'triage']
  0.0s  duplicate name: Process 'triage' already exists. Use remove_process() first or choose a different name.
  0.3s  event process.started   from digest
  0.3s  event process.started   from triage
  0.3s  event process.started   from reporter
  0.3s  event process.failed    from archiver
  2.4s  event process.failed    from reporter
  4.8s  status after 4.5 s:
         digest    running  invocations=2 failures=0 restarts=0
         triage    running  invocations=2 failures=0 restarts=0
         reporter  failed   invocations=0 failures=2 restarts=0
         archiver  failed   invocations=0 failures=0 restarts=0

The archiver failed at startup. start_all() logged AgentRuntime: failed to start archiver: Failed to connect to server 'files' ... on stderr and went on. The reporter started, failed two runs in a row, and entered FAILED. Meanwhile triage and digest kept running on their schedules, unaffected. The shared notifier delivered events from all four, each tagged with its process name in agent_id.

That is the isolation a runtime gives you: failed runs, failed starts, restart loops and budgets stay within one process. It doesn't isolate resources. All four processes share one event loop, so code that blocks it, such as a synchronous HTTP call in a lifecycle listener or a CallbackSink, stalls every agent at once. Keep your callbacks async and short.

The second part of the output, after the model provider comes back:

  4.9s  event process.started   from reporter
  4.9s  event process.stopped   from digest
  4.9s  event process.started   from digest
  7.1s  status after restarting reporter and digest, removing archiver:
         digest    running  invocations=2 failures=0 restarts=0
         triage    running  invocations=3 failures=0 restarts=0
         reporter  running  invocations=2 failures=0 restarts=0

Three different operations happened here:

• `restart_process("digest")` on a running process stops it and starts it again. You see process.stopped and process.started. Its counters, such as invocation_count, carry over.

• `stop_process("reporter")` and then `start_process("reporter")` brought the failed process back. On a FAILED process, stop() cleans up the triggers and tasks it left running, keeps the state at failed, and emits no event. Then start() builds it again. Don't use restart_process() for this in 1.4.3: on a FAILED process it skips the stop, the old trigger keeps running next to the new one, and every event fires twice. start_all() has the same gap for FAILED processes.

• `remove_process("archiver")` dropped it from the registry. remove_process() stops running processes first, but it skips FAILED ones, so call stop_process() before removing a process that failed while running. To replace a process with a new config, remove it and add it again.

For automatic restarts after failures, use the per-process restart policy instead of doing it by hand. AI agent auto-restart covers it.

Read the runtime's status

runtime.status() returns {"process_count": n, "processes": {name: status}}, where each status is the same dict as process.status(). runtime.list_processes() is the short form, with name, state and process ID. runtime.get_process(name) returns the AgentProcess itself, for example to read its dead_letters or add a lifecycle listener. AI agent lifecycle states explains every field and state.

RuntimeConfig holds every process config in one object that serializes to JSON. Build a runtime from it with from_config():

"""RuntimeConfig, from_config() and the two presets."""
import asyncio
import json

from promptise.runtime import (
    DEFAULT_DEVELOPMENT_CONFIG,
    DEFAULT_PRODUCTION_CONFIG,
    AgentRuntime,
    ProcessConfig,
    RuntimeConfig,
    TriggerConfig,
)

config = RuntimeConfig(processes={
    "triage": ProcessConfig(
        instructions="Triage new support tickets.",
        triggers=[TriggerConfig(type="cron", cron_expression="*/5 * * * *")],
    ),
    "digest": ProcessConfig(
        instructions="Write the daily digest.",
        triggers=[TriggerConfig(type="cron", cron_expression="0 7 * * *")],
        restart_policy="on_failure",
    ),
})


async def main() -> None:
    plain = AgentRuntime(config)                       # stores the config only
    print("AgentRuntime(config):", plain.list_processes())

    runtime = await AgentRuntime.from_config(config)   # registers every process
    print("from_config(config): ", [p["name"] + "/" + p["state"] for p in runtime.list_processes()])

    saved = json.dumps(config.to_dict())               # e.g. store it in a file or database
    again = RuntimeConfig.from_dict(json.loads(saved))
    print("round trip:", list(again.processes), again.processes["digest"].restart_policy)

    for name, preset in [("development", DEFAULT_DEVELOPMENT_CONFIG),
                         ("production", DEFAULT_PRODUCTION_CONFIG)]:
        print(f"{name} preset: processes={preset.processes} "
              f"distributed.enabled={preset.distributed.enabled}")


asyncio.run(main())

AgentRuntime(config): []
from_config(config):  ['triage/created', 'digest/created']
round trip: ['triage', 'digest'] on_failure
development preset: processes={} distributed.enabled=False
production preset: processes={} distributed.enabled=True

Two results deserve attention:

• `AgentRuntime(config)` only stores the config. It registered nothing. AgentRuntime.from_config(config) registers every process, and it also accepts event_bus, broker and event_notifier.

• The presets have no processes. DEFAULT_DEVELOPMENT_CONFIG and DEFAULT_PRODUCTION_CONFIG differ only in distributed.enabled, and nothing in 1.4.3 reads that flag. Using the production preset doesn't change how agents run. Build your own RuntimeConfig instead, and treat the presets as empty starting points. They are shared module-level objects, so copy one with model_copy(deep=True) before you change it.

to_dict() writes webhook hmac_secret values as **********. A config loaded back with from_dict() then fails validation until you supply the real secret, so keep secrets in environment variables and fill them in at load time.

Docker, Kubernetes and systemd stop a program by sending SIGTERM, then SIGKILL after a grace period. Python's default for SIGTERM is to exit at once, without running finally blocks or async with exits. So the runtime's cleanup never runs unless you turn SIGTERM into a normal shutdown. serve.py does that with an asyncio signal handler:

"""Run every agent in ./agents until SIGTERM or Ctrl+C, then stop them all cleanly."""
import asyncio
import logging
import signal
import time

from promptise.events import CallbackSink, EventNotifier
from promptise.runtime import AgentRuntime

T0 = time.monotonic()


def log(message: str) -> None:
    print(f"{time.monotonic() - T0:5.1f}s  {message}", flush=True)


async def on_event(event) -> None:
    log(f"event {event.event_type} from {event.agent_id}")


async def main() -> None:
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop.set)

    notifier = EventNotifier(sinks=[CallbackSink(on_event, events=[
        "process.started", "process.stopped",
    ])])
    async with AgentRuntime(event_notifier=notifier) as runtime:
        await runtime.load_directory("agents")
        await runtime.start_all()
        await stop.wait()
        busy = {n: s["invocation_count"] for n, s in runtime.status()["processes"].items()}
        log(f"signal received; completed runs so far: {busy}")
    log("all processes stopped, exiting")


logging.basicConfig(level=logging.WARNING)
asyncio.run(main())

sigterm_check.py starts serve.py, waits five seconds while runs are in flight, and sends SIGTERM:

  0.3s  event process.started from digest
  0.3s  event process.started from triage
  4.6s  signal received; completed runs so far: {'digest': 1, 'triage': 2}
  4.6s  event process.stopped from digest
  4.6s  event process.stopped from triage
  4.6s  all processes stopped, exiting
exit code: 0

The signal set the event, the program left the async with block, and stop_all() stopped both processes and drained the notifier, so both process.stopped events were delivered before the program exited with code 0. The whole shutdown took about 0.1 s.

"Graceful" here means a clean teardown, not a drain. stop() cancels runs that are in progress instead of waiting for them, and events still in a queue are not run. The event a cancelled run was working on is not dead-lettered either. If losing in-flight work matters, keep a journal and recover from it, as durable AI agents shows, and make tool calls idempotent so a repeated run is safe.

If you start the program with promptise run --prod, it converts SIGTERM into KeyboardInterrupt for you, and the same async with exit runs. Run, debug and deploy an AI agent covers that command. promptise runtime start handles Ctrl+C but not SIGTERM in 1.4.3, so use your own entry point like serve.py in containers.

multi-agent-runtime/
├── agents/
│   ├── triage.agent       # manifest, cron every 2 s (step 1)
│   └── digest.agent       # manifest with config overrides (step 1)
├── flaky_llm.py           # fake OpenAI-compatible model; down_models fails one model
├── runtime_demo.py        # load, add, fail, restart, remove (steps 2-4)
├── config_demo.py         # RuntimeConfig, from_config, presets (step 5)
├── serve.py               # production entry point with SIGTERM handling (step 6)
└── sigterm_check.py       # starts serve.py and sends SIGTERM

pip install "promptise>=1.4.3"
python runtime_demo.py
python config_demo.py
python sigterm_check.py

serve.py is the file to adapt. Point it at your agents directory, add your sinks, and run it as the container's command. To run against a real model, remove the fake server from the demos and set OPENAI_API_KEY.

Know the limits per process

A runtime has no global limits. Each process carries its own:

There are no CPU, memory or open-file limits per process, because processes are not operating system processes. If you need hard limits, put agents in separate containers. Each webhook trigger opens its own HTTP server, so every webhook in the runtime needs its own port.

Share one notifier, and alert on the right events

Pass one EventNotifier to the runtime and add sinks for what each audience needs: a LogSink for everything, a WebhookSink for process.failed or your own "gave up" event to the on-call channel. Every event carries the process name in agent_id and metadata["process_name"], so one channel can serve all agents. AI agent notifications covers sinks and signing.

Control it from outside

Wrap the runtime in the control API to list, start, stop, suspend and deploy processes over HTTP while it runs. AI agent control API covers it, including authentication. The API's restart route uses restart_process(), so restart FAILED processes with stop and start there too.

Deploy it as one container

Run serve.py as the container command, mount the journal directory on a volume, and give the container a restart policy for program crashes. Set the stop grace period above the time stop_all() needs; it is short, because runs are cancelled. Deploy an AI agent with Docker has the Dockerfile.

• One program, one machine. All processes share a Python program. The runtime has no distributed mode in 1.4.3; DistributedConfig is read by nothing.

• No resource isolation. A memory leak or CPU-heavy code in one agent affects all of them.

• No drain on shutdown. Stopping cancels runs in progress and drops queued events.

• No dependencies between processes. start_all() starts processes in registration order and doesn't wait for one to be healthy before the next.

• No built-in event bus or broker. The runtime passes through objects you provide; it doesn't create them.

Can I run several AI agents in one Python program?

Yes. Register each one as a process in a Promptise AgentRuntime, with its own model, tools and triggers, and call start_all(). They run concurrently on one asyncio event loop. Each has its own queue, workers and failure handling, so one failing agent doesn't stop the others.

What happens to the other agents when one agent fails?

Nothing. A failed run, a failed start or a process that ends up FAILED affects only that process. The other agents keep running their triggers. What isn't isolated is the program itself: if it crashes, runs out of memory or blocks its event loop, every agent is affected.

How do I restart one agent without restarting the others?

Call runtime.restart_process(name) for a running process; it stops and starts only that one. For a process in FAILED, call stop_process(name) and then start_process(name) in version 1.4.3, because restart_process() skips the cleanup and duplicates its triggers.

How do I shut down multiple AI agents gracefully?

Handle SIGTERM in your program: an asyncio signal handler sets an event, your main coroutine leaves the async with AgentRuntime() block, and the runtime calls stop_all(). That stops every process and delivers pending notifications. Runs in progress are cancelled, not finished, so use a journal if you must recover them.

Should each AI agent run in its own container?

Not necessarily. Agents that are deployed together and mostly wait on model calls run well in one runtime and one container. Use separate containers when agents need different dependencies, secrets or network access, when one is CPU-heavy, or when they must be deployed and scaled independently.

What is the difference between AgentRuntime and AgentProcess?

An AgentProcess is one long-running agent with its own triggers, queue, lifecycle and restart policy. An AgentRuntime manages many of them: it registers, starts, stops and reports on processes by name, and shares an event notifier, event bus and broker between them.

• Your first long-running agent, end to end: deploy AI agents in production.

• Restart policies for each process: AI agent auto-restart.

• Manage the runtime over HTTP: AI agent control API.

• Governance for every process: the autonomous AI agent runtime.

• Agents that work together inside one run, rather than side by side: multi-agent systems.

• Reference: runtime manager, configuration and manifests in the Promptise docs.