Kitaru vs Inngest: The runtime layer underneath your Python agents

Inngest makes the pitch: “make any code durable across an event-driven app,” and they genuinely do that. One platform takes your TypeScript, Python, or Go function, wraps it in retries and step memoization, and gives you the flow-control surface (concurrency, throttling, rate limiting, debounce, batching, prioritization) that most workflow engines leave you to wire up yourself. Add managed Inngest Cloud with SOC 2 Type II, enterprise SAML, HIPAA BAA available, and isolated branch environments for non-production branches, plus a single-binary self-host path, and you can see why a lot of app teams on Vercel or Lambda picked it. So if your durability problem is shaped like an event-driven app, Inngest is the pragmatic answer.

Kitaru is narrower on purpose. We built it for Python agents, because that’s the workload most teams we talk to are trying to ship right now. Today, it’s increasingly obvious that every company running agents in production needs the same capabilities: a durable llm() call, wait/resume, typed versioned memory, artifact lineage across runs, human/agent approval gates, and a runtime that understands the cloud underneath it. Those don’t come in the box with a general-purpose event-and-function platform. So we put them there, underneath whatever harness your team already picked rather than asking you to re-shape the app around a new platform.

What makes Kitaru unique

How the two surfaces map

import kitaru
from kitaru import checkpoint, flow

@checkpoint
def research(topic: str) -> str:
  return kitaru.llm(
      prompt=f"Research: {topic}. Return a brief.",
      model="fast",
  )

@checkpoint
def draft(brief: str) -> str:
  return kitaru.llm(
      prompt=f"Write a draft from this brief:\n{brief}",
      model="fast",
  )

@flow
def review_flow(topic: str) -> str:
  brief = research(topic)
  text = draft(brief)
  approved = kitaru.wait(
      name="approve_draft",
      question="Approve draft?",
      schema=bool,
  )
  return text if approved else "Rejected"

review_flow.run(topic="Durable agents").wait()

import datetime
import inngest

inngest_client = inngest.Inngest(app_id="agent-app")

async def call_llm(prompt: str) -> str:
  ...

@inngest_client.create_function(
  fn_id="review-flow",
  trigger=inngest.TriggerEvent(event="review.start"),
)
async def review_flow(ctx: inngest.Context) -> str:
  topic = ctx.event.data["topic"]

  async def research() -> str:
      return await call_llm(f"Research: {topic}")

  async def draft_step(brief: str) -> str:
      return await call_llm(f"Draft: {brief}")

  brief = await ctx.step.run("research", research)
  text = await ctx.step.run("draft", draft_step, brief)

  decision = await ctx.step.wait_for_event(
      "approve_draft",
      event="review.approve",
      timeout=datetime.timedelta(days=1),
  )
  return text if decision and decision.data["ok"] else "Rejected"

# Trigger:  await inngest_client.send(inngest.Event(name="review.start",   data={"topic": "..."}))
# Approve:  await inngest_client.send(inngest.Event(name="review.approve", data={"ok": True}))
# Functions register with the Inngest engine; events route through it.