अपने AI Agent को एक Nervous System दें
15 concepts, real use का लगभग 80%: senses (triggers), reflexes (durable execution), और balance (flow control).
आपने एक ऐसा agent बना लिया है जो काम करता है। लेकिन वह सिर्फ़ तभी काम करता है जब आप उसे देख रहे हों। आप Claude Code या OpenCode खोलते हैं, type करते हैं, वह reply देता है। और जैसे ही आप हट जाते हैं, वह रुक जाता है। यही gap, एक agent जिसे आप operate करते हैं और एक worker जो खुद operate करता है, इन दोनों के बीच का फ़र्क़, इस पूरे course का विषय है।
हैरानी की बात यह है कि इस gap को क्या भरता है, और वह कोई ज़्यादा smart agent नहीं है। आपके agent के पास काम करने के लिए ज़रूरी सब कुछ पहले से है: सोचने के लिए एक LLM, act करने के लिए tools और MCP servers, और जिन workflows को वह जानता है उनके लिए skills. जो उसके पास नहीं है, वह है एक nervous system. अपने ही शरीर के बारे में सोचें: आपका दिमाग़ सोचता है और आपकी muscles act करती हैं, लेकिन नीचे एक दूसरा system भी चलता है, आपकी जानकारी के बिना, आपकी heartbeat और आपके reflexes, वे signals जो सोते समय भी आपको ज़िंदा रखते हैं। आप ध्यान देना बंद कर दें तब भी आपका दिल धड़कता रहता है; agent के पास इसका कोई version नहीं है, इसलिए जैसे ही आप उसे drive करना बंद करते हैं, वह रुक जाता है। Nervous system वह connective tissue है जो खुद loop बंद करता है, बिना किसी human के हर turn चलाए: यह दुनिया को महसूस करता है और कुछ होने पर agent को जगाता है, किसी step के fail होने पर reflex से react करता है (और जब वह किसी person या किसी slow API का इंतज़ार कर रहा हो तो घंटों अपनी जगह बनाए रखता है), और जब एक साथ पाँच सौ requests आती हैं तब agent को balance में रखता है। यही वह रेखा है जो एक agent जिसे आप operate करते हैं और एक FTE जो खुद operate करता है, इन दोनों को अलग करती है। आप अपने agent को यह nervous system देते हैं; आप agent को दोबारा नहीं लिखते। यही वह एक idea है जिसके चारों ओर यह पूरा course बना है।
जो tool आपके agent को nervous system देता है उसका एक technical नाम है, durable execution engine, और हम जिसे use करते हैं उसका नाम Inngest है। ये patterns Temporal, Restate, और Dapr Agents पर भी लागू होते हैं। यह सिर्फ़ एक teaching picture नहीं है। Day AI, AI-native companies के लिए बना एक CRM, Inngest को अपने product का "nervous system" कहता है, और इस course में सिखाए गए हर हिस्से पर चलता है। Inngest का free Hobby tier शुरू करने की सबसे आसान जगह है: कोई credit card नहीं, one-command dev server, और एक dashboard जिसे आप build करते समय देख सकते हैं।
यह example जानबूझकर पतला रखा गया है: एक customer-support agent जो कुछ sample customers देखता है, एक reply draft करता है, और refund सिर्फ़ तभी issue करता है जब कोई human approve कर दे। यह जानबूझकर पतला है: मुश्किल हिस्सा agent में नहीं है, इसलिए हम उसे छोटा रखते हैं और मेहनत उसके चारों ओर के nervous system पर लगाते हैं। आप इसे यहाँ शुरू से बनाते हैं। यह पुराने Digital FTE course के साथ ideas share करता है लेकिन उसमें से कुछ भी assume नहीं करता। नीचे दिए Quick Win में environment एक बार set up करें, और Part 4 सात paste-and-watch prompts में worker बनाता है। यह Python-first है, inngest-py पर: आप अपने coding agent को plain English में direct करते हैं और वह code लिखता है। अगर आप करके सीखते हैं, तो Parts 1-3 skim करें और Part 4 पर jump करें।
किसी एक agent का task के बीच crash होना खीझ पैदा करता है। नीचे nervous system के बिना customer-facing काम संभाल रही पचास agents की workforce चलाना असंभव है: या तो आप ऐसा platform adopt करते हैं जो यह आपको देता है, या छह महीने लगाकर खुद उसका worse version बनाते हैं। चार properties इस nervous system को agents के लिए uniquely important बनाती हैं: Day AI, AI-native companies के लिए बना CRM, अपने product को इस course में सिखाए गए हर primitive पर चलाता है: durable LLM workflows, wait-for-event coordination, failure पर replay, debounce plus throttle plus concurrency, और multi-tenant fairness. उनके दो founding engineers ने अपने आप यही nervous-system picture चुना। यह production language है, curriculum branding नहीं। Agent Factory thesis उन Seven Invariants को describe करती है जिन्हें किसी भी production agent system को satisfy करना होता है। यहाँ आप जो worker बनाते हैं वह Invariant 4 (एक engine) और Invariant 5 (एक system of record, यहाँ एक छोटा audit trail) satisfy करता है। यह course दो और add करता है, plus Invariant 1 का एक हिस्सा:
एक AI agent को nervous system क्यों चाहिए (चार properties)
step.wait_for_event (Concept 15) के बिना, आप approval queue खुद बनाते हैं: database table, polling, timeout handling, audit trail. यह एक feature नहीं, पूरा project है।यह course Agent Factory thesis में कहाँ बैठता है
step.wait_for_event किसी भी platform पर इसका सबसे clean expression है: agent suspend होता है, एक human awaited event emit करता है, agent resume करता है।
15 concepts, एक नज़र में। ये उन तीन कामों पर map होते हैं जो एक nervous system करता है: senses (triggers worker को wake करते हैं), reflexes (कुछ टूटने पर durable execution उसे correct रखता है), और balance (flow control उसे load के नीचे healthy रखता है). यह first-pass version है, concept plus one-line gist. Build के दौरान जब कुछ टूटे, तो अंत में दी Quick reference में एक symptom-to-concept diagnostic है जो आपको उस concept पर वापस ले जाता है जिससे वह failure जुड़ी है।15 concepts, एक-एक line में (पूरे map के लिए expand करें)
# Concept One-line gist Senses (Triggers) दुनिया worker तक कैसे पहुँचती है 1 Events vs requests Request sync है और कोई wait करता है; event async है और दुनिया आगे बढ़ चुकी होती है। 2 Cron triggers Schedule function को wake करता है। One line: TriggerCron(cron="0 9 * * *").3 Webhook triggers Inbound HTTP payload एक named event बनता है; आपकी function name पर react करती है। 4 Idempotency and event semantics Event IDs और step names duplicate event (या retry) को no-op बनाते हैं। 5 Fan-out and sub-agent delegation One event, N subscribing functions; या one parent जो N child events fire करता है। Reflexes (Durable execution) कुछ टूटने पर worker को correct रखना 6 step.run and the durable function modelहर step.run एक checkpoint है; function steps के बीच crash होकर resume कर सकती है।7 Memoization, the mechanic underneath Completed steps re-execute होने के बजाय stored output return करते हैं। 8 step.sleep and step.wait_for_eventदोनों function को durably suspend करते हैं, एक duration के लिए या एक event के लिए। 9 Retries, error handling, dead-letter Automatic backoff retries; N tries के बाद failed run replay के लिए persist रहती है। 10 step.run for AI calls in PythonOpenAI calls को step.run में wrap करें; step.ai.infer inference offload करता है (step.ai.wrap सिर्फ़ TypeScript के लिए है)।Balance (Flow control) load के नीचे worker को healthy रखना 11 Concurrency and throttling concurrency active runs cap करता है; throttle starts-per-second cap करता है।12 Priority and fairness Priority queue order तय करती है; per-key concurrency हर tenant को fair share देती है। 13 Batching सस्ते bulk work के लिए events को एक batched function call में accumulate करें। 14 Replay and bulk cancellation Failed runs को new code के साथ replay करें; जिन runs की अब ज़रूरत नहीं उन्हें bulk-cancel करें। 15 HITL gates with step.wait_for_eventFunction तब तक suspend रहती है जब तक कोई human approve न करे, फिर decision के साथ resume करती है।
Prerequisites. चार चीज़ें, और बाक़ी मामले में course अकेला खड़ा रहता है (Part 4 अपना worker शुरू से बनाता है)।
- आप एक coding agent drive कर सकते हैं। Claude Code या OpenCode, installed और authenticated. Plan mode, rules files, read-first-then-write workflow: अगर यह rhythm जाना-पहचाना लगता है, तो आप calibrated हैं। अगर नहीं, तो Agentic Coding Crash Course इसे cover करता है।
- आपके पास एक
OPENAI_API_KEYहै (या कोई दूसरा model key जिसे आपका coding agent use कर सके) और worker के Postgres system of record के लिए एक Neon account है। Worker एक real model चलाता है और अपने customers तथा audit trail को Neon में read और write करता है। Neon free है (कोई card नहीं), और setup के दौरान आप एक browser click से उसे authorize करते हैं; अगर account नहीं है तो neon.com पर लगभग एक minute में sign up करें। Inngest dev server को खुद किसी account की ज़रूरत नहीं है।- आपके पास Node.js 20+ available है, भले worker Python हो। Inngest dev server एक Node CLI के रूप में distribute होता है (
npx inngest-cli@latest dev)।- आपके पास "event-driven" vs "request/response" का एक working mental model है। अगर "दुनिया एक event fire करती है और zero, one, या many functions उस पर react करती हैं" जाना-पहचाना लगता है, तो आप calibrated हैं। अगर नहीं, तो Concept 1 आपको shape देता है।
From Agent to Digital FTE कर चुके हैं? तो आपके पास wrap करने के लिए एक richer worker है; Part 4 के अंत में एक callout nervous system को उस पर point करता है। यह एक bonus है, gate नहीं।
इस page को first pass में कैसे पढ़ें, plus जिन terms से सामना होगा उनकी glossary
First pass. "Done when" या "What to watch" labeled किसी भी चीज़ को expand करें: runnable behavior जिसके against अपनी predictions check करें। Part 4 में आप first read में load-bearing snippets skim कर सकते हैं; हर snippet के आसपास का prose बताता है कि वह layer क्या करती है, और जब आप build करते हैं तब आपका agent code लिखता है। "Try with AI" blocks optional extension prompts हैं। Pass one का goal nervous-system model, उसकी तीन layers, आपके दिमाग़ में बैठाना है; pass two, keyboard पर हाथ रखकर, वह जगह है जहाँ आप build करते हैं। हर concept एक Predict (आगे पढ़ने से पहले एक answer पर commit करें) या एक Quick check (अभी पढ़े नियम को test करें) पर खत्म होता है; दोनों आपको रुकने के लिए हैं, grade करने के लिए नहीं।
Glossary (हर term को उस context में भी समझाया गया है जहाँ वह पहली बार आती है):
- Production Worker: एक AI agent जिसके चारों ओर एक nervous system है: senses जो उसे wake करते हैं (triggers), reflexes जो failures survive करते हैं (durable execution), और balance जो उसे load के नीचे scale करता है (flow control)।
- Event: एक named, immutable message जो describe करता है कि कुछ हुआ। Example:
{"name": "customer/email.received", "data": {"customer_id": "..."}}. Trigger surface. - Inngest function:
@inngest_client.create_functionसे decorated एक Python function, जो triggers और steps declare करती है। Durable work की unit. - Step: एक Inngest function के अंदर work की unit जो
ctx.step.run(),ctx.step.sleep(),ctx.step.wait_for_event(), याctx.step.ai.infer()में wrapped होती है। हर step independently retried और memoized होता है। - Memoization: जब function crash होकर restart होती है, Inngest function code को ऊपर से re-run करता है लेकिन किसी भी ऐसे
step.runके लिए stored outputs return करता है जिसका result पहले से cached है। Function बिना काम दोहराए वहीं तक catch up कर जाती है जहाँ टूटी थी। - Flow control: Per-function policies:
concurrency(max active runs),throttle(max starts per second),priority(queue order),batch_events(invoke करने से पहले accumulate करना)। - HITL (Human In The Loop): एक function आगे बढ़ने से पहले human approval या input के लिए pause करती है।
step.wait_for_eventइसका primitive है। - Replay: Failed runs को bug fix के बाद current code पर ऊपर से fresh runs के रूप में फिर चलाना (एक run के अंदर के automatic retry से अलग, जो memo से resume होता है)। Dashboard का Rerun button।
- Dev server: Inngest का local dev environment
npx inngest-cli@latest devसे। Dashboardhttp://127.0.0.1:8288पर; MCP endpoint/mcpपर।
मई 2026 तक current. पूरा Part 4 build एक live Inngest dev server और एक real model के against end-to-end चलाया गया था, inngest 0.5.18, openai-agents 0.17.3, fastapi 0.136.3, Python 3.12, और Inngest CLI पर। Part 4 का हर snippet उसी working build से है, memory से लिखा नहीं गया। यह course जो architecture सिखाता है वह SDK बदलने पर नहीं बदलता; SDK इस साल का उसका interface है। अगर किसी live docs page और इस page के बीच किसी syntax detail पर मतभेद हो, तो docs जीतते हैं: अपने versions pin करें, और build करते समय Inngest Python quick start तथा OpenAI Agents SDK docs देखें।
जिन sections में Claude Code और OpenCode अलग होते हैं वहाँ एक switcher है; एक चुनें और page visits के across sync रहता है।
पंद्रह-minute quick win: base set up करें, और reflex देखें
15 concepts पढ़ने से पहले जो समझाते हैं कि यह architecture क्यों काम करता है, वह environment set up करें जिसमें पूरा course चलता है और एक durable function को एक crash survive करते देखें। यह setup आप एक बार करते हैं; Part 4 ठीक इसी base पर customer-support worker बनाता है। अंत तक आपके पास होगा:
- base आपके coding agent में खुला हुआ, Skills installed, और तीन MCP servers wired (Neon, Context7, और dev-server
inngest-dev), - एक fresh Neon database जिसमें दो tables हैं,
customersऔरaudit_log, जिन्हें आपने MCP से बनाया और console में देखा, और जिसकाDATABASE_URLबाद में worker के लिए.envमें लिखा गया, - एक tiny durable function (एक
step.run, एकstep.sleep, एक FastAPI host) जो Inngest dev server के against चल रही है, - एक run जिसे आपने trigger किया और sleep पर zero compute consume किए suspend होते देखा,
- और एक run जिसे आपने जानबूझकर तोड़ा, फिर Inngest को retry करते देखा, जो पहले से completed step को memo से return करता रहा जबकि सिर्फ़ broken step फिर से execute हुआ।
वह आख़िरी beat ही पूरे course का सार है, छोटे रूप में: वह reflex जिसे आप अपनी आँखों से देख सकते हैं, एक step fail होता है और system उस काम को दोहराए बिना recover कर लेता है जो वह पहले ही finish कर चुका था। यह Part 4 का worked example नहीं है (पूरा Worker, सात prompts); यह एक बैठक है। इसे करें, फिर concepts के लिए वापस आएँ।
एक Production Worker दो processes हैं जो साथ-साथ चलती हैं, और इन्हें अलग रखना ही mental model है: एक Python function host (आपका code, जो function को Inngest को serve करता है) और Inngest dev server (वह nervous system जो runs trigger करता है, steps memoize करता है, और आपको dashboard दिखाता है). आपका coding agent दोनों को wire करता है, वे Skills install करता है जो उसे Inngest patterns सिखाती हैं, और inngest-dev MCP से dev server से बात करता है।
एक और boundary matter करती है, और यह वही है जो Digital FTE course ने खींची थी। आपका worker अपने customers और अपना audit trail एक Neon Postgres database में रखता है, और उस database को छूने के दो अलग तरीके हैं। आपका coding agent उसे build और inspect करने के लिए Neon MCP use करता है: tables बनाना, rows पढ़ना, connection string खींचना, सब development time पर plain English में। आपका worker उसे runtime पर read और write करने के लिए अपना ही Postgres connection (DATABASE_URL) use करता है। Worker कभी Neon MCP call नहीं करता, और Neon के अपने docs इसकी वजह साफ़-साफ़ बताते हैं: MCP server development और inspection के लिए है, किसी running app में कभी wire नहीं किया जाता। Neon एक OAuth click से free है; Inngest dev server को किसी account की ज़रूरत बिल्कुल नहीं।
Base पाएँ और उसे खोलें
Base download करें और folder को अपने coding agent में खोलें। Agent setup खुद करता है, नीचे दिए prompts से। आप यह एक बार set up करते हैं: ai-agent-nervous-system/ पूरे course के लिए आपका folder है, Quick Win और Part 4 दोनों के लिए। आप कभी फिर download या unzip नहीं करते।
ai-agent-nervous-system-base.zip download करें
cd ai-agent-nervous-system
claude
यह base एक capable general agent मानकर चलता है (Claude Code, या Claude Sonnet या Opus, GPT-5, या इसी तरह का कुछ चलाता OpenCode). एक छोटा model build prompt पर drift करेगा; अगर उसका पहला plan specific के बजाय vague लगे, तो आगे बढ़ने से पहले एक stronger model पर switch करें।
Base prep करें (~3 min)
Base अपने rules AGENTS.md में और अपनी MCP wiring में ship करता है; Skills, आपका key, और Neon authorization इसके बाद आते हैं। अपने agent से खुद को set up कराएँ। यह paste करें:
Read AGENTS.md, then get this base ready: install the Skills it lists for whichever agent you are, copy
.env.exampleto.envfor me, and tell me exactly what you need from me to bring the Neon and Context7 MCP servers online.
Watch for: agent चार Inngest Skills और neon-postgres Skill install करता हुआ (आप install runs और Installed confirmations देखते हैं), .env बनाता हुआ, फिर आपसे दो चीज़ें माँगता हुआ: .env में paste करने के लिए आपका OPENAI_API_KEY, और Neon को OAuth से authorize करने के लिए एक browser click. Neon free है; अगर अभी account नहीं है, तो neon.com पर लगभग एक minute में sign up करें, या authorization screen पर ही एक बना लें। INNGEST_DEV=1 पहले से .env में है, इसलिए SDK बिना signing key के local dev mode में चलता है। Install और wiring पूरी होने पर, agent आपको dev server start करने (अगला step) और फिर उसे restart करने को कहता है, क्योंकि नई Skills और inngest-dev MCP session के बीच load नहीं होते।
Done when: Skills installed हैं, .env में आपका key है, Context7 reachable है, और Neon authorized है। inngest-dev MCP तब online आता है जब dev server चल रहा हो, जो अगला step है।
Dev server start करें, और confirm करें कि agent उस तक पहुँच सकता है (~2 min)
यह course दो boundaries add करता है जिन तक आपका agent MCP से पहुँचता है: एक Neon database जिसे वह build और inspect करता है, और running dev server जिसे वह events भेजता है और देखता है। इसलिए कुछ भी build करने से पहले, दोनों को up करें और confirm करें कि वे live हैं।
Inngest dev server को उसके अपने terminal में start करें (यह एक Node CLI है; इसे चलता छोड़ दें):
npx inngest-cli@latest dev
Dashboard http://127.0.0.1:8288 पर आता है, और dev server अपना MCP endpoint /mcp पर expose करता है। अब अपने coding agent को restart करें (exit करके ai-agent-nervous-system folder में फिर launch करें) ताकि नई installed Skills और inngest-dev MCP दोनों load हों। फिर यह paste करें:
List the Neon tools and the inngest-dev tools you can see.
Watch for: दो real lists. Neon tools (project बनाना, SQL चलाना, tables describe करना, connection string लाना, वग़ैरह) database पर आपके agent का हाथ हैं। inngest-dev tools (list_functions, send_event, invoke_function, get_run_status, और बाक़ी) running dev server पर उसका हाथ हैं। नीचे सब कुछ इन दोनों पर सवार है।
Gate open: reply में real Neon tool names और real inngest-dev tool names दोनों दिखें। अगर Neon tools missing हैं: OAuth पूरा नहीं हुआ; prep step से Neon authorization दोबारा करें। अगर inngest-dev tools missing हैं: dev server नहीं चल रहा (उसे start करें), या आपने restart skip किया (exit करें, इस folder में फिर launch करें, फिर पूछें)।
Store बनाएँ, और उसका connection string ले लें (~3 min)
अब worker का system of record Neon MCP से बनाएँ, फिर worker को वह एक चीज़ दें जिसकी उसे बाद में पहुँचने के लिए ज़रूरत होगी: एक connection string. आप Part 4 में जो worker बनाते हैं वह अपने customers यहीं पढ़ता है और अपना audit trail यहीं लिखता है। यह paste करें:
Paste this to your coding agent. Plan first; execute on approval.
On a fresh Neon project, create two tables:
customers(id, email, tier) andaudit_log(a record of every action the worker takes). Then call the Neon tool that returns the connection string and write that URL into my.envasDATABASE_URL. Use the Neon tools for all of it; don't write SQL for me to run.
Watch for: agent project और दोनों tables बनाने के लिए Neon MCP tools call करता हुआ (आप वे tool calls देखते हैं, आपके typed SQL नहीं), फिर .env में DATABASE_URL लिखता हुआ। वह string ही handoff है: Neon MCP ने store provision किया, और आपका worker उस string का use करेगा, MCP server का नहीं।
Done when: एक fresh Neon project मौजूद है जिसमें एक customers table और एक audit_log table है, और .env में एक DATABASE_URL है। console.neon.tech खोलें, वह project चुनें जो agent ने अभी बनाया, और Tables खोलें: वहाँ customers और audit_log बैठे हैं, अभी खाली। जब worker चलेगा तो D0 में आप rows आते देखेंगे। (एक table बस एक spreadsheet है: हर row एक चीज़, हर column एक detail।)
पहली durable function बनाएँ, और उसे dashboard से drive करें (~3 min)
अब सबसे छोटी durable function बनाएँ, उन Skills को use करते हुए जो आपने अभी install कीं। Inngest Skills अपने examples में TypeScript-first हैं, इसलिए आपका agent उनसे patterns लेता है (एक step क्या होता है, एक durable function की shape कैसी होती है) और exact Python signatures docs से confirm करता है (dev-server MCP का grep_docs/read_doc, या Context7), memory से नहीं। यह paste करें:
Using the Inngest Skills, write one tiny Inngest durable function (call it
greet-customer, triggered by ademo/greetevent) that composes a greeting in onestep.run, sleeps fifteen seconds withstep.sleep, then composes a farewell in a secondstep.runand returns both. Serve it from a FastAPI host in local dev mode, and start the host on port 8000 with auto-reload on, so edits I make later are picked up without a manual restart.
वह shape जो यह लिखता है, ताकि देखकर आप पहचान लें: function plain async def है, दो step.run calls उस काम को wrap करते हैं जिसे memoize होना चाहिए, और उनके बीच का step.sleep run को durably suspend करता है (process sleep के दौरान crash, restart, या redeploy हो सकता है; timer fire होने पर run अगली line पर resume करता है). Agent के code में confirm करने लायक एक detail: Inngest client is_production=False से construct होता है, या वह .env में पहले से मौजूद INNGEST_DEV=1 पढ़ता है। इनमें से एक के बिना, SDK चुपचाप Cloud पर default करता है और आपकी function locally कभी register नहीं होती।
Done when: function host port 8000 पर चल रहा है, और dev server (पिछले step से पहले से चल रहा) ने उसे auto-discover कर लिया। http://127.0.0.1:8288 खोलें, Functions click करें, और greet-customer listed है। बाक़ी आप browser से drive करते हैं।
इसे trigger करें, और एक step को zero compute पर sleep करते देखें (आप drive करते हैं)
Trigger event भेजें। सबसे simple path dashboard है: http://127.0.0.1:8288 में, Events click करें, फिर Send event, यह paste करें, और Send click करें:
{
"name": "demo/greet",
"data": { "name": "Sara" }
}
(Agent में ही रहना चाहते हैं? उससे MCP पर event भेजने को कहें: "Send a demo/greet event with name Sara using the inngest-dev send_event tool." दोनों तरीक़ों से वही run शुरू होता है।)
Runs click करें और नई run खोलें। पहला step complete होता है; sleep step एक resume time के साथ Sleeping दिखाता है। आपके code में कुछ नहीं चल रहा, host terminal idle है, और यही point है: एक durable wait zero compute में होता है। पंद्रह seconds बाद run अपने आप resume होती है, farewell step complete होता है, और status पलटकर Completed हो जाता है। Output panel returned dict दिखाता है।
एक step को तोड़ें, और retry को वह काम skip करते देखें जो वह पहले ही कर चुका था (payoff)
अब एक step को जानबूझकर fail कराएँ, ताकि आप memoization को retry के पार completed काम ले जाते देख सकें। यह अपने agent को paste करें:
Make the farewell step raise an error on purpose, so I can watch a run fail. Keep everything else the same.
वही demo/greet event फिर भेजें, फिर run खोलें और उसका trace पढ़ें। यहाँ payoff है, और यह इसी एक failing run में है: greeting step एक completed attempt दिखाता है, और farewell step कई Attempts दिखाता है, हर एक backoff के साथ retried (Inngest कई attempts पर default करता है) इससे पहले कि run Failed पर land करे। उस attempt count के मतलब के साथ ठहरें: completed greeting step एक बार के लिए pay किया गया, हर retry पर एक बार नहीं। यह durable execution है जिसे आप अपनी आँखों से देख सकते हैं। क्यों completed step re-run होने के बजाय तुरंत return होता है, वह mechanic आप Concept 7 में मिलेंगे; अभी के लिए, बस उसे होते देखें।
(यह dev-server build कोई अलग "memoized" badge नहीं दिखाता। Memo ही attempt count है: completed step एक attempt पर बैठा हुआ जबकि broken step चढ़ता हुआ, बिल्कुल यही है कि "memo से return हुआ, re-run नहीं" यहाँ कैसा दिखता है।)
अब इसे fix करें:
Now revert the farewell step to the working version.
Host auto-reload होता है (यही --reload ने आपको दिया; अगर आपने उसे skip किया, तो host को हाथ से restart करें)। एक fresh demo/greet event भेजें और पूरी function अब fixed code पर साफ़ चलकर Completed हो जाती है। Recovery के बारे में एक honest note, क्योंकि यह लोगों को परेशान करता है: dashboard का Rerun button आपके current code के साथ ऊपर से एक बिल्कुल नया run शुरू करता है, हर step शुरू से re-execute होता हुआ। वह incident recovery के लिए सही tool है (एक bad deploy ने runs के एक batch को तोड़ दिया; आप एक fix ship करते हैं और उन्हें rerun करते हैं), लेकिन वह memo-preserving resume नहीं है। Memo-preserving resume वह automatic retry है जो आपने अभी failing run के अंदर देखा, जहाँ completed step अपनी जगह बना रहा।
आपने अभी पूरा course environment set up कर लिया और nervous system को अपनी आँखों से काम करते देखा: Skills installed हैं, आपका Neon store provision हो चुका है और .env में DATABASE_URL है, dev-server MCP live है, और आपने एक durable function चलाई, एक step को बिना compute consume किए sleep करते देखा, फिर एक step तोड़ा और automatic retry को completed step memo से return करते देखा जबकि सिर्फ़ broken step फिर चला। यही वह architecture है जिसके बारे में यह course है। बाक़ी course इसे scale up करता है: असली senses (cron, webhook, fan-out), stronger reflexes (step.run के अंदर agent invocation), load के नीचे असली balance, और वह human-approval gate जो "agent इसे गड़बड़ कर सकता है" को "agent draft करता है, एक human approve करता है, action issue होता है" में बदलता है।
अगर कुछ काम नहीं किया, तो चार problems लगभग सब कुछ cover करते हैं:
- Dev server function host तक नहीं पहुँच पा रहा: confirm करें कि host port 8000 पर चल रहा है।
- Client Cloud mode में है: agent ने
is_production=Falseछोड़ दिया और.envमेंINNGEST_DEV=1नहीं है, इसलिए functions locally कभी register नहीं होतीं। उससे कोई एक set कराएँ (एक explicitis_productionvalue env var पर जीतती है)। - Function dashboard से missing है: host reload नहीं हुआ; उसे restart करें।
- एक run बिना error और बिना progress के hang हो जाती है: एक de-synced host चुपचाप stall होता है; host और dev server दोनों को साथ restart करें, और एक host को एक dev server के against चलाएँ। (एक subtle वजह: अगर
:8288लिया हुआ था और dev server8289+पर आया, तोinngest-devMCP URL को re-point करना काफ़ी नहीं; host अब भी:8288से बात करता है। Host परINNGEST_BASE_URL=http://127.0.0.1:<port>set करें ताकि वह dev server के पीछे नए port पर जाए।)
अगर इनमें से कोई हिट हो, तो universal recovery move यहाँ भी काम करता है: "Something didn't work. Read the error, tell me in plain language what you see, and propose one fix I can approve."
आपने क्या बनाया, और यह कहाँ बढ़ता है
Environment set up हो गया: base खुला है, Skills installed हैं, तीनों MCP servers wired हैं (Neon, Context7, inngest-dev), आपके Neon store में customers और audit_log tables हैं और .env में DATABASE_URL है, और dev server चल रहा है। आपने वह एक idea भी अपनी आँखों से देखा जिस पर पूरा course टिका है, durable execution का reflex. Part 4 इसी base पर, इसी folder में customer-support worker बनाता है: यह उन्हीं customers को पढ़ता है और उन्हीं audit rows को लिखता है, फिर पूरी चीज़ को पूरे nervous system में wrap करता है, एक real event trigger, एक daily cron जो fan out करता है, flow control, और refunds पर durable human-approval gate. Part 4 इस step.run-and-step.sleep skeleton को एक ऐसे worker में scale करता है जो आपके Neon store पर real काम करता है। अगर यह Quick Win काम कर गया, तो आगे के concepts समझाते हैं कि हर piece ऐसा shape क्यों है।
Part 1: senses, दुनिया worker तक कैसे पहुँचती है
जिस AI agent को आप हाथ से call करते हैं वह तब चलता है जब आप उसे call करें। एक असली Production Worker के पास senses होते हैं: वह तब चलता है जब दुनिया उस तक पहुँचती है। एक customer email करता है, एक webhook आता है, एक cron रोज़ 09:00 पर fire होता है, एक दूसरा worker काम hand off करता है। इनमें से हर एक अंदर आता हुआ एक signal है, और एक trigger वह है जिससे agent उसे महसूस करता है। Part 1 के पाँच concepts वही senses हैं: event-driven mental model, दुनिया के अंदर पहुँचने के तीन तरीके (cron, webhook, event), वे semantics जो double-processing रोकती हैं, और वे fan-out patterns जिनसे एक signal कई workers को wake कर सकता है।
Concept 1: Events vs requests, durable mental model shift
इस course में आगे जो कुछ भी है वह एक mental shift पर टिका है: requests से events तक।
एक request एक synchronous conversation है। कोई call करता है; आप handle करते हैं; आप return करते हैं; वे continue करते हैं। एक connection खुला रहता है; एक human या service wait कर रही होती है। अगर आप crash करते हैं, caller को error मिलता है। जिस agent से आप prompt पर chat करते हैं वह एक request है: आपने type किया, उसने stream back किया, conversation आपके terminal session की थी।
एक event एक asynchronous message है। दुनिया में कुछ हुआ (एक customer ने sign up किया, एक email आया, एक payment clear हुई), और originator उस fact का एक named record emit करता है। Zero, one, या many functions उस event पर independently react करती हैं। कोई connection खुला नहीं रहता। Originator को नहीं पता कौन listen कर रहा है, वह results के लिए wait नहीं करता, और blocked नहीं होता। दुनिया आगे बढ़ चुकी होती है।
# A request: I'm here, waiting, blocking
result = await agent.handle_customer_message(text=user_input)
print(result) # I unblock when the agent finishes
# An event: I fire-and-forget
await inngest_client.send(events=[
inngest.Event(
name="customer/email.received",
data={"customer_id": "c-4429", "body": email_body, "subject": subject},
),
])
# I return immediately. Somewhere else, one or more Inngest
# functions react to this event on their own schedule.
यह shift छोटा लगता है। है नहीं। एक बार आप events में सोचने लगें, तो durability और scale लगभग free में निकल आते हैं, क्योंकि:
- Producer consumer से slow नहीं हो सकता (email-receiver agent के reply draft करने का wait नहीं करता)।
- Consumer crash और restart कर सकता है बिना काम खोए (event durably stored है; Inngest उसे re-deliver करता है)।
- Producers को बदले बिना नए consumers add किए जा सकते हैं (एक दूसरी function, मान लें एक analytics counter,
customer/email.receivedपर subscribe कर सकती है बिना email-receiver को पता चले)। - Backpressure एक code change नहीं, एक flow-control policy बन जाता है (Inngest concurrency cap करता है; producer fire करता रहता है; events queue होते हैं)।
Predict. आपका customer-support Worker एक email का response देने में 8 seconds लेता है: agent की reasoning के लिए तीन seconds, दो MCP tool calls के लिए चार seconds, database write के लिए एक second. Peak load पर आपको per minute 50 emails मिलते हैं। अगर आप request model use करते हैं (email parser agent के finish होने तक block करता है), तो आपके email parser तक कितने parallel HTTP connections imply होते हैं? अगर आप event model use करते हैं (email parser एक event fire करके तुरंत return करता है), तो कितने? Confidence 1-5.
Answer: request model को लगभग 7 concurrent parsers चाहिए (50/min × 8 seconds = ~6.7 parallel handlers, plus headroom). Event model को एक parser चाहिए (वह event fire करके ~10ms में return करता है; event queue 50/min spike absorb करती है; Inngest functions queue को उतनी concurrency पर consume करती हैं जितनी आप allow करते हैं)। Event model production rate को consumption rate से decouple करता है। यह सिर्फ़ एक scaling fact नहीं; यह एक architectural fact है। Event "दुनिया में क्या हुआ" और "Worker उसके बारे में क्या करता है" के बीच एक durable boundary बन जाता है। Consumer को बीच में crash कराएँ और event फिर भी retry के लिए मौजूद है। तीन और consumer types add करें और producer notice नहीं करता। Events वह तरीका हैं जिससे आप काम की timing own करना बंद करते हैं।Try with AI
Walk me through three scenarios. For each, classify it as REQUEST-MODEL
or EVENT-MODEL, and explain which one fits better:
A) A user clicks "Submit refund request" in the support portal and
expects to see "Refund issued: $30" within 2 seconds.
B) A nightly cron job at 02:00 runs a customer-health-check across
all 5,000 customers and writes a report to Slack.
C) A customer sends an email to support@; we want a draft response
ready within 60 seconds for the on-call agent to review and send.
For each, name (a) what the human's expectation of timing is and
(b) what failure looks like if the model crashes mid-execution.
Concept 2: Cron triggers, वह काम जो इसलिए चलता है क्योंकि समय बीता
सबसे simple trigger clock है। एक Production Worker के कई काम बाहरी events के reactions नहीं होते; वे scheduled काम होते हैं: daily health reports, weekly cleanups, hourly recalculations. Inngest का cron trigger code की एक line है।
import inngest
@inngest_client.create_function(
fn_id="daily-customer-health-check",
trigger=inngest.TriggerCron(cron="0 9 * * *"), # 09:00 every day, UTC
)
async def daily_health_check(ctx: inngest.Context) -> dict[str, int]:
"""Run a customer-health pass for every Pro/Enterprise customer."""
customers = await ctx.step.run("fetch-pro-customers", fetch_pro_customer_ids)
# fan out: one event per customer, one Worker run per event
await ctx.step.run("fan-out", fan_out_per_customer_events, customers)
return {"customers_scheduled": len(customers)}
तीन चीज़ें notice करें:
-
Schedule बस standard cron syntax है।
0 9 * * *हर दिन 09:00 UTC है;*/15 * * * *हर 15 minutes है;0 9 * * 1Mondays at 09:00 है। Inngest cron को UTC में evaluate करता है; अगर आपको कोई दूसरा timezone चाहिए, वह एक function parameter है, अलग concept नहीं। -
Function अभी भी
ctx.step.runuse करती है। Cron-triggered हो या event-triggered, function shape identical है। Steps वैसे ही काम करते हैं। Durability वैसे ही काम करती है। Flow control वैसे ही काम करता है। Trigger सिर्फ़ यह है कि function कैसे start होती है। -
Cron का output एक regular Inngest function run है। यह dashboard में दिखता है, इसका एक run ID होता है, एक trace होता है, replay support करता है। अगर आपका Monday-morning cron run step 3 पर fail होता है, Tuesday का cron normally चलेगा और Monday की failure bug fix के बाद replay के लिए available रहेगी।
अगर cron fire होते समय आपकी service down हो तो क्या होता है? यही वह सवाल है जो एक durable scheduler को एक नाज़ुक scheduler से अलग करता है। Schedule fire होते ही Inngest के cron runs durably record हो जाते हैं; अगर आपका function endpoint unreachable है, Inngest backoff के साथ retry करता है जब तक succeed न हो या retry ceiling पर न पहुँचे। 09:00 पर fire हुआ cron इसलिए "miss" नहीं होता कि आपका deploy 09:00 पर rolling था; run wait करता है, आप deploy finish करते हैं, run complete होता है। Development में cron triggers की एक quirk जानने लायक है: local dev server crons सिर्फ़ तभी fire करता है जब वह चल रहा हो। Production उन्हें Inngest के infrastructure पर चलाता है, जो हमेशा चल रहा होता है।
Quick check. तीन claims. हर एक को True या False mark करें। (a) अगर एक cron function चलने में 45 minutes लेती है और हर 15 minutes scheduled है, तो किसी भी समय तीन concurrent instances चल रही होंगी। (b) आप एक cron-triggered function के अंदर
step.sleepuse करके काम को पूरे दिन में spread कर सकते हैं। (c) एक cron-triggered function को testing के लिए dashboard से manually भी invoke किया जा सकता है।
Answers: (a) Depends on concurrency policy: default रूप से Inngest overlapping runs को queue करेगा; अगर आप concurrency=1 set करते हैं तो वे serialize होंगी; अगर concurrency=10 set करते हैं तो parallelize होंगी। Default sane है। (b) True, और यह "daily काम को घंटों में spread करके load smooth करने" का एक common pattern है। (c) True: Inngest dashboard testing के लिए किसी भी function को on demand invoke करने देता है, उसके trigger से independent.Try with AI
With my AI coding assistant connected to the Inngest dev server MCP,
write a cron-triggered Inngest function in Python that:
1. Runs every Monday at 09:00 UTC.
2. Queries the audit_log table for all conversations resolved in the
prior week (status='resolved' in that window).
3. Computes per-agent metrics: total conversations resolved, average
resolution time, count of escalations, count of refunds issued.
4. Returns the metrics as a JSON object.
After you write the function, use the MCP's `invoke_function` tool to
test it manually (instead of waiting for Monday). Confirm the audit
SQL is correct by using `grep_docs` to search Inngest's docs for
"step.run" examples.
Concept 3: Webhook triggers, जब बाहरी दुनिया अंदर call करती है
दूसरा trigger surface HTTP है। एक external system (Stripe, आपका email provider, एक customer-portal form, एक GitHub webhook) आपके Worker को call करना चाहता है। Inngest के बिना, आपको यह सब करना पड़ता: एक HTTPS endpoint खड़ा करना, payload parse करना, source validate करना, एक queue में write करना, queue से consume करने वाला एक worker लिखना, retries handle करना, idempotency handle करना, telemetry ship करना। इनमें से हर एक infrastructure work का एक हफ़्ता है।
Inngest के साथ, endpoint दिया हुआ है। आप Inngest dashboard में https://inn.gs/e/<your-key> जैसी URL के साथ एक webhook configure करते हैं, Stripe (या जो भी हो) को उस URL पर point करते हैं, और webhook payload आपके event stream में एक event बन जाता है. Matching event-name trigger वाली कोई भी function अब fire होती है।
@inngest_client.create_function(
fn_id="handle-stripe-refund-failed",
trigger=inngest.TriggerEvent(event="stripe/charge.refund.failed"),
)
async def on_refund_failed(ctx: inngest.Context) -> dict[str, str]:
"""Triggered by Stripe webhook → Inngest event → this function."""
charge_id = ctx.event.data["charge_id"]
customer_id = ctx.event.data["customer_id"]
# Look up which support ticket originated this refund
ticket = await ctx.step.run(
"find-ticket-for-refund", lookup_ticket_by_charge, charge_id,
)
# Wake the customer-support Worker with the full context
await ctx.step.run(
"notify-support-agent",
notify_support_agent_of_refund_failure,
ticket_id=ticket["id"], charge_id=charge_id,
)
return {"ticket": ticket["id"], "action": "notified"}
Flow: Stripe एक charge refund करने में fail होता है → Stripe Inngest webhook URL पर POST करता है → Inngest stripe/charge.refund.failed named एक event create करता है → ऊपर वाली function (जो उस event name से match करती है) fire होती है → function steps use करके ticket lookup करती है और support agent को notify करती है। HTTP plumbing में से कुछ भी आपको लिखना नहीं है। कोई endpoint नहीं, कोई parser नहीं, कोई queue नहीं, कोई consumer नहीं।
दो adjacent patterns नाम देने लायक हैं:
- Generic JSON webhooks. अगर source कोई known vendor नहीं है, तो आप किसी भी JSON-emitting service को उसी तरह के endpoint पर point करते हैं और event name चुनते हैं। Slash-namespaced names (
vendor/event.subtype) convention हैं; कोई इसे enforce नहीं करता, लेकिन follow करने पर dashboard साफ़-सुथरा sort होता है। - Webhook transforms. अगर incoming payload उस shape से match नहीं करता जो आप चाहते हैं, Inngest आपको एक "transform" function define करने देता है जो receipt time पर server-side चलती है और event को आपके event stream में enter होने से पहले reshape करती है। इससे आपका function code provider-specific fields से साफ़ रहता है।
Predict. एक Stripe webhook ठीक उसी millisecond पर
stripe/charge.refund.failedfire करता है जब आपका customer-support Worker भीinngest_client.sendcall करके एक अलग event namedcustomer/refund.investigation_neededemit कर रहा है। दोनों events system में एक साथ arrive होते हैं; ऊपर वाली function सिर्फ़ Stripe event पर trigger होती है। Function एक बार चलेगी या दो बार? Confidence 1-5.
Answer: एक बार. Function सिर्फ़ stripe/charge.refund.failed पर trigger होने के लिए registered है; customer/refund.investigation_needed event का एक अलग name है और वह एक different function से match करता है (या किसी function से नहीं, अगर आपने एक नहीं लिखी)। एक event का name उसकी routing key है। अलग names वाले दो events कभी accidentally same function trigger नहीं करते, भले वे एक ही instant पर arrive हों। यह एक वजह है कि naming discipline matter करती है: एक event name में typo (customer/email_received vs customer/email.received) का मतलब function कभी fire नहीं होती, और symptom silent होता है। Inngest का dashboard इसे catch करने में मदद करता है: unmatched events एक अलग stream में appear होते हैं जिसे आप audit कर सकते हैं।Try with AI
I need to handle three webhook sources for my customer-support Worker:
A) Stripe: refund failed, charge disputed
B) Postmark (email service): bounced email, complaint
C) My internal admin UI: manual "investigate this ticket" button
For each, decide:
1. What event names you'd use (vendor/event.subtype format).
2. Whether the function reacting to it should run synchronously (the
caller is waiting) or asynchronously (fire and continue).
3. Whether you'd write a webhook transform to reshape the payload, or
consume it raw.
Then write the Inngest function for the Stripe refund-failed case in
Python, using the MCP's grep_docs to find the current syntax for
TriggerEvent and the dev-server MCP's send_event tool to test it.
Concept 4: Idempotency and event semantics, वही event दो बार fire होना
Webhooks exactly-once नहीं होते। वे at-least-once होते हैं: sender को acknowledgment न मिले तो वह retry करता है। Networks packets drop करते हैं, services restart होती हैं, आपका endpoint timeout होता है और sender retry करता है भले आप actually succeed कर चुके हों। Idempotency के बिना, हर webhook system अंततः किसी को double-bill, double-email, या double-refund करता है। यह कोई theoretical concern नहीं है; यह event systems में सबसे common production bug है।
Defense की दो layers, दोनों Inngest में built in.
Layer 1: source पर Event ID seed. जब आप खुद एक event भेजते हैं (एक webhook receive करने के बजाय), तो आप एक idempotency key attach कर सकते हैं:
await inngest_client.send(events=[
inngest.Event(
name="customer/refund.requested",
data={"order_id": "o-4429", "amount_cents": 5000},
id=f"refund-request-{order_id}-{request_timestamp}", # idempotency key
),
])
अगर same id वाला एक दूसरा event dedup window (default 24 hours) के अंदर भेजा जाता है, Inngest duplicate को drop कर देता है। वही logical event, वही id, सिर्फ़ एक function run.
Layer 2: Step-level idempotency. एक function के अंदर, हर step.run अपने name से identified होता है। अगर एक function step 3 और step 4 के बीच crash होती है, retry function code को ऊपर से re-run करता है, लेकिन steps 1, 2, और 3 के लिए Inngest step body को re-execute किए बिना stored outputs return करता है। Step 4 पहली बार normally चलता है। यही एक function को "durable" बनाता है: completed steps के side effects retry पर फिर नहीं होते।
@inngest_client.create_function(
fn_id="issue-customer-refund",
trigger=inngest.TriggerEvent(event="customer/refund.requested"),
)
async def issue_refund(ctx: inngest.Context) -> dict[str, str]:
# Step 1: look up the order. If the function retries, this returns
# the SAME order data it computed the first time, from Inngest's memo.
order = await ctx.step.run(
"lookup-order", lookup_order_by_id, ctx.event.data["order_id"],
)
# Step 2: call Stripe. If the function retries AFTER this step
# succeeded, the Stripe call does NOT happen again. The refund is
# issued exactly once even if the function runs three times.
refund = await ctx.step.run(
"issue-stripe-refund", call_stripe_refund_api,
charge_id=order["stripe_charge_id"],
amount=ctx.event.data["amount_cents"],
)
# Step 3: write the audit row. Same property: runs at most once.
await ctx.step.run(
"audit-refund", write_audit_refund_issued,
order_id=order["id"], refund=refund,
)
return {"refund_id": refund["id"]}
अगर यह function step 3 के दौरान crash होती है, retry step 1 में re-enter करता है (cached order data मिलता है, कोई DB call नहीं), step 2 में re-enter करता है (cached refund data मिलता है, कोई Stripe call नहीं), step 3 real में चलता है, return करता है। Customer का card एक बार charge होता है, भले function तीन बार चली। यही killer feature है। यही Inngest को retry loop वाली एक queue से qualitatively अलग बनाता है।
Inngest की memoization आपको function के perspective से exactly-once step completion देती है: एक बार step.run किसी step को successful record कर दे, वह re-execute नहीं होगा। लेकिन एक narrow window है। अगर आपके step की body Stripe call करती है (side effect Stripe के servers पर होता है), फिर Inngest के result record करने से पहले crash हो जाती है, retry Stripe को फिर call करेगा। Inngest के perspective से step "complete नहीं हुआ।" Stripe के perspective से charge पहले ही हो गया। Production-grade pattern Inngest step memoization plus provider-level idempotency keys है: Stripe का Idempotency-Key header, Postmark का MessageID reuse, आपके अपने MCP server का idempotency contract. step.run और provider idempotency keys को complements की तरह treat करें, substitutes की तरह नहीं: step.run आपकी function की internal logic को exactly-once रखता है; provider की idempotency key external side effect को exactly-once रखती है।
Quick check. True या false. (a)
step.runकिसी step को idempotent सिर्फ़ तभी बनाता है जब अंदर वाली function भी idempotent हो। (b) Dedup window के बाहर एक duplicate ID वाला event एक new event की तरह treat होगा। (c) अगरstep.runmid-execution fail होता है (step का code एक exception throw करता है), Inngest failure store करता है और next attempt पर step retry करता है, prior steps को re-run किए बिना।
Answers: (a) False: step.run step के invocation को idempotent बनाता है (success पर वह at most once चलेगा), लेकिन अगर अंदर की function non-idempotent है (जैसे Stripe call), तो at-most-once guarantee वही है जो आप चाहते हैं। पूरा point यही है कि आपको Stripe-calling को खुद idempotent नहीं बनाना पड़ता। (b) True: Inngest की dedup window default रूप से 24 hours है; उस window के बाद same ID वाले events new treat होते हैं। (c) True: automatic retry memoized है; Inngest जानता है step 3 attempt 1 पर fail हुआ और attempt 2 पर सिर्फ़ step 3 retry करता है। Prior successful steps re-execute नहीं होते। (यह within-run retry है, dashboard का Replay button नहीं, जो एक fresh run है, Concept 14।)Try with AI
Here are three scenarios. For each, decide: idempotency PROBLEM or
NO PROBLEM, and if it's a problem, what's the fix:
A) Stripe sends the same charge.refund.failed webhook three times
in 90 seconds (because their first two attempts timed out at
your endpoint). Your function emails the customer.
B) A customer clicks "Issue refund" three times because the page
was slow. Your function calls Stripe and writes audit_log.
C) Your nightly cron at 09:00 sends a customer-health-check event
to each Pro customer. If two crons fire at the same time (a deploy
bug), what happens?
For each problem case, propose ONE specific fix: event ID seed
inside the function, idempotency key in inngest_client.send, or
function-level deduplication on the trigger.
Concept 5: Fan-out and sub-agent delegation, एक event कई Workers
अक्सर एक single event को कई जगह काम trigger करना होता है। Stripe charge.refund.failed event को शायद यह करना हो: support agent को notify करना, audit में write करना, customer का risk score update करना, finance ops को alert करना, Slack पर post करना। पाँच reactions, सब independent, सब एक event से।
Inngest pattern: same event पर कई functions subscribe करें। कोई fan-out code नहीं; बस same TriggerEvent के साथ कई @inngest_client.create_function decorators. हर function independently चलती है, उसके अपने retries होते हैं, उसका अपना step trace होता है, और वह दूसरों से independently fail होती है।
@inngest_client.create_function(
fn_id="refund-failed-notify-support",
trigger=inngest.TriggerEvent(event="stripe/charge.refund.failed"),
)
async def notify_support(ctx: inngest.Context) -> dict[str, str]:
# ... runs the customer-support Worker to draft a response ...
return {"status": "drafted"}
@inngest_client.create_function(
fn_id="refund-failed-update-risk-score",
trigger=inngest.TriggerEvent(event="stripe/charge.refund.failed"),
)
async def update_risk_score(ctx: inngest.Context) -> dict[str, float]:
# ... runs the risk-scoring Worker ...
return {"new_risk_score": 0.42}
@inngest_client.create_function(
fn_id="refund-failed-post-slack",
trigger=inngest.TriggerEvent(event="stripe/charge.refund.failed"),
)
async def post_to_slack(ctx: inngest.Context) -> None:
# ... posts a Slack notification ...
return None
एक Stripe webhook आता है। Inngest एक event create करता है। तीन functions fire होती हैं, हर एक अपने run में। अगर post_to_slack fail होता है क्योंकि Slack down है, बाक़ी दो unaffected रहती हैं और normally complete होती हैं। Failed run dashboard में replay के लिए बैठी रहती है जब Slack recover हो। यही multi-Worker coordination का core है, और यही architectural pattern आपकी future manager layer (एक later course) scale पर compose करेगी।
दूसरा fan-out pattern: parent-fires-N-children. कभी fan-out dynamic होता है। आपके daily cron को हर Pro customer के लिए एक customer-health event fire करना है, जो week के हिसाब से 500 या 5,000 हो सकता है। Parent function N events भेजती है:
from datetime import date
async def fan_out_per_customer_events(
customers: list[str],
) -> int:
events = [
inngest.Event(
name="customer/health_check.requested",
data={"customer_id": cid},
id=f"daily-health-{cid}-{date.today().isoformat()}", # idempotency
)
for cid in customers
]
await inngest_client.send(events=events)
return len(events)
एक single send call में 5,000 events भेजे जाते हैं। 5,000 function runs fire होती हैं, हर एक के पास अपना customer_id है, हर एक isolated है, हर एक independently retryable है। Flow control (Concept 11) cap करता है कि कितने concurrently चलें ताकि आप अपने downstream APIs न पिघलाएँ। Cron function seconds में return करती है; fan-out उसी rate पर चलता है जो Inngest की flow-control policies allow करती हैं।
Sub-agent delegation fan-out का एक special case है। एक Worker run के अंदर, आप await inngest_client.send(...) call करके sub-tasks दूसरे Worker types को delegate कर सकते हैं। Parent children का wait नहीं करता जब तक वह explicitly step.invoke use करके उन्हें synchronously न चलाए और उनके results collect न करे।
Predict. आपके पास तीन functions हैं जो सभी
customer/email.receivedसे triggered हैं: reply draft करने वाला customer-support agent (15 seconds), एक analytics counter (50ms), और एक "VIP detector" जो check करता है कि customer high-value है या नहीं (200ms). जब एक email आता है, हर एक के लिए user-visible latency कैसी दिखती है? तीन options: (a) तीनों जुड़कर ~15 seconds; (b) तीनों parallel चलते हैं, total latency ~15 seconds (सबसे slow); (c) हर एक independently चलता है, कोई shared latency बिल्कुल नहीं। Confidence 1-5.
Answer: (c). हर function अपना run है, अपने process slot में। Customer-support agent analytics counter को block नहीं करता; VIP detector agent को block नहीं करता। बाहर से, किसी particular function की latency बस उसी function का अपना time है। कोई function कभी किसी sibling function का wait नहीं करती। यही वजह है कि fan-out scale करता है: consumers isolated हैं। अगर agent crash होता है, analytics counter unaffected है।Try with AI
Design the fan-out architecture for these three scenarios. For each,
sketch the event names and the functions that subscribe:
A) New customer signs up. Need to: send welcome email, create
Stripe customer, post to Slack #new-customers, write to
audit_log, schedule a 7-day follow-up.
B) Customer support email arrives. Need to: draft a reply (agent),
detect sentiment, check if VIP, update customer's "last contact"
timestamp, attach to the right ticket thread.
C) Daily cron at 09:00 needs to run customer-health-check on
~5,000 Pro customers. Each check takes ~30 seconds. We want
the whole batch to complete by 11:00 (a 2-hour window).
For each, decide: how many event types, how many subscriber
functions, what the idempotency story is, and one specific failure
mode this design protects against.
Part 2: reflexes, कुछ टूटने पर क्या होता है
Senses worker को wake करते हैं। Reflexes worker को उसके बाद आने वाली चीज़ें survive कराते हैं। एक worker एक agent call करता है, agent कुछ tools call करता है, tools एक database और एक payment API और एक model call करते हैं: एक ही turn में कई external calls, जिनमें से कोई भी fail हो सकता है। Durability के बिना, turn के बीच एक transient failure पूरी flow को ऊपर से restart कर देती है। एक reflex automatic है: यह तेज़ी से act करता है, बिना agent के mind के तय किए। यही वह है जो durable execution आपको देता है। Durability वह property है जो कहती है: जब कुछ mid-execution fail होता है, पहले से completed काम completed रहता है, और execution वहीं से resume होता है जहाँ टूटा था। Inngest इसे एक primitive (step.run) और उसके नीचे एक memoization mechanic से deliver करता है। Part 2 दोनों explain करता है, plus time-based variants (step.sleep, step.wait_for_event), retry semantics, और step.ai primitives.
First-pass compression note. अगर आप scan कर रहे हैं, load-bearing concepts 6 (
step.run) और 7 (memoization) हैं। Concepts 8-10 उन्हीं पर build करते हैं। 6 और 7 ध्यान से पढ़ें; ये दोनों आपके दिमाग़ में बैठ जाएँ तो बाक़ी तेज़ पढ़ेंगे।
Concept 6: step.run and the durable function model
एक normal Python function एक बार चलती है, ऊपर से नीचे। अगर वह आधे रास्ते crash होती है, तो आप ऊपर से शुरू करते हैं। अगर वह crash होने से पहले तीन API calls करती है, तो अगली attempt वे तीनों calls फिर करती है, और उनके लिए pay करती है, और शायद किसी को फिर double-charge करती है।
एक Inngest function durable है। जिस operation को आप checkpoint कराना चाहते हैं वह step.run(name, fn, ...) में wrapped होता है। Function हर attempt पर अभी भी ऊपर से नीचे चलती है, लेकिन जो steps पहले ही complete हो चुके हैं वे re-execute होने के बजाय अपने stored outputs return करते हैं. Function वहीं तक "catch up" कर जाती है जहाँ टूटी थी, फिर आगे continue करती है।
@inngest_client.create_function(
fn_id="customer-support-conversation",
trigger=inngest.TriggerEvent(event="customer/email.received"),
)
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
customer_id = ctx.event.data["customer_id"]
# Step 1: load the customer record (one DB call)
customer = await ctx.step.run(
"load-customer", load_customer_by_id, customer_id,
)
# Step 2: load the conversation thread (one DB call)
thread = await ctx.step.run(
"load-thread", load_thread_for_customer, customer_id,
)
# Step 3: run the OpenAI Agents SDK agent (your worker)
response = await ctx.step.run(
"run-agent",
run_customer_support_agent,
customer=customer,
thread=thread,
email_body=ctx.event.data["body"],
)
# Step 4: write the draft reply to the database
await ctx.step.run(
"save-draft-reply", save_reply,
customer_id=customer_id, text=response.draft,
)
# Step 5: notify the on-call human reviewer via Slack
await ctx.step.run(
"notify-reviewer", post_slack_for_review, response=response,
)
return {"status": "drafted", "reviewer_notified": True}
पाँच steps। हर एक independently checkpointed है।
यहाँ durability आपको क्या देती है, तीन failure scenarios में:
-
Scenario A: agent step एक timeout throw करता है। अगर agent call को
step.runwrap न करे, तो इस function का अगला retry customer फिर load करता है, thread फिर load करता है, और agent को शुरू से फिर चलाता है, जो काम agent पहले ही आंशिक रूप से कर चुका था उसके लिए OpenAI tokens दो बार pay करता हुआ।step.runके साथ, customer और thread loads memoized हैं (steps 1-2 re-execute नहीं होते); सिर्फ़ step 3 retry होता है। Inngest के automatic retries transient OpenAI errors को आपके code के बिना जाने handle कर लेते हैं। -
Scenario B: function process step 3 और step 4 के बीच kill हो जाता है (एक deploy roll out हुआ, एक node restart हुआ, container OOM हो गया)। Durability के बिना, agent का response खो जाता है और customer का email तब तक unanswered रहता है जब तक कोई ध्यान न दे। Durability के साथ, function restart के बाद resume होती है: steps 1, 2, 3 milliseconds में अपने stored outputs return करते हैं, step 4 real में चलता है, step 5 real में चलता है, customer को drafted reply मिल जाता है।
-
Scenario C: Slack step 5 पर एक 503 return करता है।
step.runके बिना, या तो आप काम खो देते या Slack call के लिए ख़ासतौर पर retry-and-backoff logic हाथ से लिखते।step.runके साथ, Inngest step 5 को exponential backoff के साथ retry करता है जब तक Slack recover न हो; इस बीच steps 1-4 completed रहते हैं और re-execute नहीं होते। Draft reply पहले से database में है; notification ही एकमात्र चीज़ है जो pending है।
आप कोई retry loops, कोई "क्या मैं यह पहले ही कर चुका" checks, कोई state machines नहीं लिखते। State machine है ही step.run calls का sequence. हर step एक node है; हर transition durable है।
step.run का एक नियम। जो function step.run को pass की जाती है उसे अपने inputs को देखते हुए deterministic होना चाहिए: उसे same arguments के साथ दो बार call करने पर same result आना चाहिए।
- Pure functions के लिए यह automatic है।
- Idempotent API calls (Stripe का
idempotency_key, आपके अपने MCP server tools) के लिए यह automatic है। - "एक random ID generate करना" या "default temperature पर एक LLM call करना" जैसी चीज़ों के लिए इसमें सावधानी चाहिए (एक retry original attempt से अलग output पैदा कर सकता है, जो कभी-कभी matter करता है)।
जब operation deterministic नहीं होता, तो आप उसे deterministic बनाते हैं: एक seed pass करें, step के बाहर random value pre-generate करें, या स्वीकारें कि retry original से अलग हो सकता है (एक agent response के लिए अक्सर ठीक है)।
Quick check. True या false. (a) Function body हर retry पर ऊपर से re-execute होती है, जिसमें
step.runcalls के बाहर के सारे imports और variable assignments शामिल हैं। (b) अगर एक step complete होने में 30 seconds लेता है, और function 25 seconds में crash होती है, तो retry उस step को second 25 से continue करता है। (c)step.runके outputs Inngest के infrastructure में store होते हैं, आपके application में नहीं।
Answers: (a) True, और यही वजह है कि आप काम को step.run के अंदर रखते हैं। step.run के बाहर का code हर retry पर फिर चलता है; अंदर का code per attempt एक बार चलता है और success पर memoized होता है। (b) False: step.run atomic unit है; अगर एक step interrupt होता है, retry पूरे step को re-run करता है। अगर आपका step इतना लंबा है कि उसे restart नहीं किया जा सकता, तो आप उसे छोटे steps में तोड़ते हैं। (c) True: step output store Inngest का हिस्सा है, आपके DB का नहीं। यही वजह है कि आप runs replay कर सकते हैं भले आपका database schema बदल गया हो।Try with AI
With my AI coding assistant connected to the Inngest dev server MCP,
shape a customer-support worker into an Inngest durable function.
Take a Runner.run call that processes a customer email and wrap each
of these inside its own step.run:
1. Load the customer record
2. Load the related conversation thread
3. Run the agent (the OpenAI Agents SDK Runner)
4. Persist the draft reply
5. Notify the on-call reviewer
Use grep_docs to find the current Python SDK syntax. Use
invoke_function to test it with a synthetic email payload. Then
deliberately raise an exception in step 4 and use get_run_status
to confirm steps 1-3 don't re-execute on retry.
Concept 7: Memoization, resumability के नीचे का mechanic
Concept 6 ने कहा "जो steps पहले ही complete हो चुके हैं वे re-execute होने के बजाय अपने stored outputs return करते हैं।" वह mechanism memoization है और उस mechanic को समझना worth है, क्योंकि हर दूसरा Inngest primitive इसे use करता है।
जब आप await ctx.step.run("load-customer", load_customer_by_id, "c-4429") call करते हैं, पहली attempt पर तीन चीज़ें होती हैं:
- Inngest अपना memo store check करता है: "क्या इस run में step
load-customerके लिए कोई stored result है?" नहीं है। - Function
load_customer_by_id("c-4429")चलती है। यह{"id": "c-4429", "tier": "pro", ...}return करती है। - Inngest उस result को memo store में लिखता है,
(run_id, step_name="load-customer")से keyed. फिर वह result आपके code को return करता है।
अगर function step 3 के बाद crash होती है और Inngest retry करता है, तो दूसरी attempt पर function body ऊपर से re-run होती है। जब execution उसी line तक पहुँचता है, तो तीन अलग चीज़ें होती हैं:
- Inngest अपना memo store check करता है: "क्या इस run में step
load-customerके लिए कोई stored result है?" हाँ, यह attempt 1 पर store हुआ था। - Function
load_customer_by_id("c-4429")नहीं चलती. DB call नहीं होती। - Inngest stored result milliseconds में आपके code को return करता है।
यही वजह है कि retries सस्ते हैं: महँगा काम पहले से cached है। यही वजह है कि durability correct है: महँगा काम दो बार नहीं होता। और यही वजह है कि "function body ऊपर से नीचे फिर चलती है" ठीक है, भले यह बेकार लगे: steps के अंदर का काम actually फिर नहीं चलता; सिर्फ़ steps के बीच का orchestration code चलता है.
वह implication जो नए users को हैरान करता है। step.run के बाहर का code हर attempt पर चलता है। अगर आप यह करते हैं:
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
# ANTI-PATTERN: this runs on every retry. Don't do this.
expensive_thing: dict = await fetch_expensive_data(ctx.event.data["id"])
await ctx.step.run("do-something", do_something_with, expensive_thing)
return {"status": "done"}
fetch_expensive_data हर retry पर चलता है। अगर एक call $0.10 का है और function 5 बार retry होती है, तो आपने अभी same data पाँच बार fetch करने में $0.50 खर्च कर दिए। Fix यह है कि महँगी चीज़ को उसके अपने step में wrap करें:
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
expensive_thing: dict = await ctx.step.run(
"fetch-expensive-data", fetch_expensive_data, ctx.event.data["id"],
)
await ctx.step.run("do-something", do_something_with, expensive_thing)
return {"status": "done"}
अब fetch_expensive_data memoized है; retries इसके लिए फिर pay नहीं करते।
Step name ही memo key है। यही वजह है कि एक function के अंदर step names unique होने चाहिए। अगर आपके पास same function में दो step.run("load-customer", ...) calls हैं, Inngest दोनों calls के लिए पहले वाले का stored output return करेगा। यह लगभग कभी वह नहीं होता जो आप चाहते हैं। अगर आपके पास एक loop है जो एक step को N बार call करता है, उन्हें uniquely name करें (step.run(f"load-customer-{i}", ...)) ताकि हर iteration का अपना memo slot हो।
Predict. आपकी function में तीन steps हैं। Step 1 (
load-customer) DB calls में $0.01 का है और 100ms लेता है। Step 2 (run-agent) OpenAI tokens में $0.20 का है और 12 seconds लेता है। Step 3 (save-draft) DB calls में $0.005 का है और 50ms लेता है। Step 2, OpenAI rate limits के कारण 30% बार fail होता है; Inngest backoff के साथ retry करता है। (a) तीनों कोstep.runमें wrap करने और (b) सिर्फ़ step 2 कोstep.runमें wrap करने के बीच cost का फ़र्क़ क्या है? Confidence 1-5.
Answer: (a) के साथ, एक single retry आपको सिर्फ़ step 2 का cost देता है ($0.20). Customer और save-draft memoized हैं; वे re-execute नहीं होते। (b) के साथ, हर retry आपको steps 1 और 3 plus step 2 का cost देता है: per retry $0.215. 30% retry rate के साथ हज़ार emails पर, यह pure waste में लगभग $4.50 का फ़र्क़ है, plus यह पता लगाने की operational complexity कि step 3 दो बार चलने पर आंशिक रूप से क्या लिखा गया। जो कुछ भी आप re-executed नहीं देखना चाहते उसे step.run में wrap करें। mechanic समझ लेने के बाद यह optional नहीं है।Try with AI
With my AI coding assistant: review the Inngest function we built
in Concept 6's Try-with-AI and identify any code BETWEEN step.run
calls that should be wrapped in its own step but isn't. Common
candidates:
- Computed values (timestamps, IDs, formatting) that we want to be
stable across retries
- Calls to logging or metrics services
- Reads from Redis, environment variables, secret managers
Then propose a refactor that moves each of these into its own step
with a meaningful name. For each, explain whether the side effect
is one you want to happen once (use step.run) or every retry
(leave it outside).
Concept 8: step.sleep and step.wait_for_event, समय के पार durability
कुछ काम को wait करना पड़ता है। एक welcome-email pipeline तुरंत एक email भेजता है, फिर तीन दिन wait करता है, फिर एक follow-up भेजता है। एक refund-investigation को एक human के approve करने का wait करना होता है। एक trial-conversion flow 7 दिनों के अंदर "user upgraded to paid" के लिए देखता है और जो वह देखता है उसके हिसाब से एक अलग email भेजता है।
एक normal Python function में, "तीन दिन wait करो" का मतलब है एक process को तीन दिन खुला रखना। यह नामुमकिन है: आपका process restart होता है, आपका hosting आपको 72 घंटे के idle compute के लिए bill करता है, आपका timer खो जाता है। Inngest में, "तीन दिन wait करो" एक line है:
from datetime import timedelta
@inngest_client.create_function(
fn_id="trial-welcome-series",
trigger=inngest.TriggerEvent(event="user/trial.started"),
)
async def welcome_series(ctx: inngest.Context) -> dict[str, str]:
user_id = ctx.event.data["user_id"]
await ctx.step.run("send-welcome-email", send_welcome_email, user_id)
# Wait three days. The function gets paged out of memory. Nothing
# is consuming compute. Three days later, Inngest pages it back in
# and resumes execution at the next line.
await ctx.step.sleep("wait-three-days", timedelta(days=3))
await ctx.step.run("send-followup", send_followup_email, user_id)
return {"status": "completed"}
step.sleep durable है, यह nervous system का आराम है। Function suspend होती है; Inngest resume time store करता है; wait के दौरान कुछ भी compute consume नहीं करता; function सही समय पर resume होती है, सारे prior step outputs अभी भी memoized. step.sleep (और step.sleep_until) paid plans पर एक साल तक, free Hobby plan पर सात दिन तक wait कर सकता है (Inngest usage limits). सात-दिन का Hobby ceiling इस course के हर sleep के लिए काफ़ी चौड़ा है।
ज़्यादा powerful sibling step.wait_for_event है। समय के लिए wait करने के बजाय, किसी दूसरे event के लिए wait करें। Function तब तक suspend रहती है जब तक एक matching event न आए, या जब तक आपके set किए timeout की मियाद खत्म न हो। यही Inngest को HITL (Concept 15) और inter-agent coordination patterns का सबसे clean expression बनाता है:
@inngest_client.create_function(
fn_id="refund-with-approval",
trigger=inngest.TriggerEvent(event="customer/refund.requested"),
)
async def refund_with_approval(ctx: inngest.Context) -> dict[str, str]:
request = ctx.event.data
request_id = request["request_id"]
# If amount is over $100, require approval before issuing
if request["amount_cents"] >= 10_000:
# Notify a human via Slack/email/whatever
await ctx.step.run("notify-approver", notify_human_approver, request)
# Wait for an approval event. Up to 24 hours; expires otherwise.
approval = await ctx.step.wait_for_event(
"wait-for-approval",
event="refund/approval.decided",
timeout=timedelta(hours=24),
if_exp=f"async.data.request_id == '{request_id}'",
)
if approval is None or not approval.data.get("approved"):
return {"status": "rejected_or_timeout"}
# Either it was under $100, or it was approved
refund = await ctx.step.run(
"issue-stripe-refund", call_stripe_refund_api, request,
)
return {"status": "issued", "refund_id": refund["id"]}
क्या हो रहा है:
- Function
wait_for_eventतक पहुँचती है। वह suspend होती है। Zero compute consume होता है। - एक human Slack notification देखता है, आपके admin UI में "Approve" click करता है, आपका UI
inngest_client.send(events=[Event(name="refund/approval.decided", data={"request_id": "...", "approved": True})])call करता है। - Inngest event को waiting function से match करता है (
if_expयह सुनिश्चित करता है कि सिर्फ़ इस request_id के events match हों) और function कोapprovalreturn value के रूप में event के साथ resume करता है। - Function refund step तक continue करती है। Stripe refund human के approve करने के बाद होता है।
step.sleep और step.wait_for_event ऐसे timeouts हैं जिनके लिए आप pay नहीं करते। Function आपके code में synchronous दिखती है ("तीन दिन wait करो, फिर email भेजो"), लेकिन runtime semantics async और durable हैं। यह उन दो चीज़ों में से एक है जिनके लिए Inngest मशहूर है (दूसरी durable retries)। इसके बिना, alternative एक queue plus एक state machine plus एक database plus एक poller है, और आप तीन के बजाय हज़ार lines लिखते।
Quick check. तीन claims. हर एक को True या False mark करें। (a) अगर
step.sleep30 दिनों के लिए set है और आपकी service उन 30 दिनों में पाँच बार redeploy होती है, तो एक paid plan पर sleep बिना रुके continue रहता है। (b) अगरstep.wait_for_eventtimeout होता है, function एक exception raise करती है। (c) same function में दोstep.wait_for_eventcalls एक साथ same event के लिए wait कर सकती हैं।
Answers: (a) एक paid plan पर True: sleeps Inngest के infrastructure में store होते हैं, आपकी service की memory में नहीं, इसलिए redeploys उन्हें खोते नहीं। Tier ceiling नोट करें: एक 30-दिन का sleep paid plan पर ठीक है लेकिन free Hobby plan के सात-दिन sleep cap से अधिक है। (b) False: timeout पर, wait_for_event None return करता है। आपका code उसे check करता है और तय करता है कि क्या करना है (rejection, escalation, default-approval, जो भी policy हो)। (c) True, पर संदिग्ध: एक matching event आने पर दोनों fire होंगी। अगर दोनों wait_for_event calls के अलग if_exp filters हैं, यह ठीक है। अगर वे identical हैं, तो आप शायद एक refactor opportunity देख रहे हैं।Try with AI
Build a delayed-investigation flow with my AI coding assistant.
Specification:
1. Triggered by event 'customer/refund.failed'.
2. Immediately notify the on-call human via Slack with the refund
details and a "Investigate" button.
3. Wait for the human to click the button (which fires
'customer/refund.investigation_started') for up to 4 hours.
4. If the click arrives in time: run the agent to draft an
investigation summary.
5. If 4 hours pass without a click: escalate to a senior reviewer
by firing 'customer/refund.escalated'.
Use the dev-server MCP's send_event tool to simulate the
human-click event during testing. Use get_run_status to inspect
how the suspended function shows up in the dashboard. Before
writing, use list_docs to scan the Inngest documentation tree
for the right page on wait_for_event semantics, then
read_doc on the page you find to get the exact syntax for
the if_exp filter expression.
Concept 9: Retries, error handling, dead-letter
यह reflex को क़रीब से देखना है। Default रूप से, Inngest failed steps को retry करता है। Defaults समझदार हैं: ~4 retries exponential backoff के साथ, attempts के बीच कुछ seconds से कुछ minutes तक। आख़िरी retry fail होने के बाद, run एक failed state में जाती है और inspection तथा (optionally) replay के लिए वहीं रहती है। आप इसे per function tune कर सकते हैं: retries=10, retries=0 (बिल्कुल retry न करें), specific exception types जिन्हें retry नहीं होना चाहिए।
@inngest_client.create_function(
fn_id="charge-customer",
trigger=inngest.TriggerEvent(event="order/checkout.completed"),
retries=2, # only retry twice; this involves Stripe; don't keep hammering
)
async def charge_customer(ctx: inngest.Context) -> dict[str, str]:
try:
charge = await ctx.step.run(
"call-stripe", call_stripe_charge, ctx.event.data,
)
return {"status": "charged", "charge_id": charge["id"]}
except StripeCardDeclinedError as e:
# A declined card is not a transient failure. Don't retry.
# Mark the order as failed in our database and emit an event
# for the dunning flow.
await ctx.step.run(
"mark-failed", mark_order_failed,
ctx.event.data["order_id"], reason=str(e),
)
await ctx.step.run(
"emit-dunning-event", emit_dunning, ctx.event.data["order_id"],
)
return {"status": "card_declined"}
तीन patterns matter करते हैं।
Pattern 1: Transient vs permanent failures. Inngest default रूप से सब कुछ retry करता है, लेकिन कुछ errors transient नहीं होतीं। Stripe से एक card-declined error retry पर फिर declined होगी। आपके downstream API से एक 401-unauthorized सिर्फ़ wait करने से 200 नहीं बन जाएगी। आपकी function को इन्हें ख़ासतौर पर catch और handle करना चाहिए: अपने DB में write करें, एक downstream event emit करें, साफ़-सुथरा return करें, ताकि वे hopeless attempts पर retry budget बर्बाद न करें। Inngest का NonRetriableError Inngest को explicitly बताता है कि एक thrown exception के लिए retries skip कर दे।
Pattern 2: Step-level vs function-level errors. एक step जो throw करता है retry होता है। Step-level retries खत्म होने के बाद, function fail होती है। कभी आप चाहते हैं कि एक function एक failing step को survive कर ले: failure log करे, काम को "partial" mark करे, continue करे। step.run को try/except में wrap करें। Step को फिर भी अपने retries मिलते हैं; अगर सारे retries fail होते हैं, exception आपके catch block तक propagate होता है, जहाँ आप तय कर सकते हैं कि क्या करना है।
Pattern 3: Dead-letter and replay. जब एक function पूरी तरह fail होती है, वह गायब नहीं होती। वह Inngest dashboard के "failed runs" view में जाती है, पूरे trace, सारे step outputs, exception, और एक Replay button के साथ। एक bug fix ship करने के बाद, आप failed runs replay कर सकते हैं: हर एक fixed code पर ऊपर से फिर चलती है (एक fresh run, memo-preserving resume नहीं, वह distinction Concept 14 है)। यह traditional queues का "dead-letter queue" pattern है, बस आप dead-letter handler नहीं लिखते। आप बस bug fix करके replay करते हैं, side-effecting steps को idempotent रखते हुए ताकि एक re-run double-act न करे।
Predict. आपकी function step 2 में Stripe call करती है और step 4 में आपका customer data service. Stripe step 2 की पहली attempt पर 503 (service unavailable, transient) return करता है। Step 2 exponential backoff के साथ 4 बार retry होता है (~1s, ~2s, ~5s, ~12s); 4थी retry पर Stripe वापस आ जाता है, charge succeed होता है। अब step 4 चलता है, और data service एक 500 के साथ down है। Inngest पूरी function retry करता है, या सिर्फ़ step 4? कितनी बार? Confidence 1-5.
Answer: सिर्फ़ step 4, और उसे अपना retry budget मिलता है। Steps retries share नहीं करते। Step 2 की चार retries step 4 की retries से independent हैं। Inngest step 4 को retry करेगा (default ~4 बार) और अगर MCP server वापस आता है, step 4 complete होता है, और function succeed होती है। Step 2 का Stripe charge फिर issue नहीं होता, क्योंकि step 2 का output उसकी successful retry के बाद memoized था। Customer एक बार charge होता है भले function ने retries में 20 seconds बिताए।Try with AI
With my AI coding assistant: extend the customer-support Worker
function from Concept 6 with explicit retry and failure handling.
Specification:
1. The OpenAI Agents SDK call should retry 3 times on transient
failures (rate limit, timeout), but NOT retry on a content-policy
refusal from the model.
2. The Slack notification should retry up to 10 times (Slack is
often flaky; don't lose the notification).
3. The Postgres write should retry once; if it fails again, log the
failure and continue (don't fail the whole function over a
transient DB blip).
For each step, decide what's transient vs permanent and structure
the try/except accordingly. Use grep_docs to find the Python SDK's
NonRetriableError equivalent.
Concept 10: Python में AI calls के लिए step.run (step.ai.wrap सिर्फ़ TypeScript के लिए है)
Concepts 6-9 किसी भी side-effecting code के लिए काम करते हैं: DB writes, API calls, file writes, agent invocations. Inngest AI-specific step primitives भी ship करता है जो उन patterns को handle करते हैं जिनके LLM calls शिकार होते हैं: rate-limit retries, prompts और responses में observability, और (optionally) inference proxying जो serverless compute costs कम करती है।
एक ज़रूरी Python-vs-TypeScript note पहले ही। Inngest के
step.aimodule में दो methods हैं, और उनका language support अलग है।step.ai.infer()TypeScript और Python दोनों में available है (Python SDK v0.5+): यह inference को Inngest के infrastructure पर offload करता है और call को trace करता है।step.ai.wrap()सिर्फ़ TypeScript है: आज इसका कोई Python equivalent नहीं है। Python projects (जैसे इस course के Worker) के लिए, एक OpenAI Agents SDK call को wrap करने का correct patternctx.step.run(...)है, जो आपको पहले से wrapped step के inputs और outputs की full durability, retries, और observability देता है। आपको बस वह LLM-specific prompt/response telemetry नहीं मिलती जो TypeScriptstep.ai.wrapadd करता है। (AI Inference docs के against मई 2026 तक verified.)
Python में OpenAI calls के लिए step.run (recommended pattern). आपकी function OpenAI call ctx.step.run("name", fn, ...) के अंदर करती है। Inngest step के inputs और outputs को trace करता है (जो arguments आपने pass किए और जो return हुआ), transient failures पर retry करता है, और result memoize करता है ताकि बाद के steps की retries OpenAI cost फिर न दें। Prompt और response dashboard में step के input/output के रूप में record होते हैं:
from openai import AsyncOpenAI
oai = AsyncOpenAI()
async def call_openai_summary(thread_text: str) -> str:
"""A normal async function. Inngest doesn't care that this is an LLM call."""
response = await oai.chat.completions.create(
model="gpt-5",
messages=[
{"role": "system", "content": "Summarize this support thread in 3 sentences."},
{"role": "user", "content": thread_text},
],
)
return response.choices[0].message.content
@inngest_client.create_function(
fn_id="summarize-customer-thread",
trigger=inngest.TriggerEvent(event="customer/thread.summary_requested"),
)
async def summarize_thread(ctx: inngest.Context) -> dict[str, str]:
thread: list = await ctx.step.run(
"load-thread", load_thread, ctx.event.data["thread_id"],
)
# The OpenAI call is wrapped in step.run. Inngest sees this as a step:
# the inputs (formatted thread text) are recorded, the output (summary
# string) is recorded, the call is memoized on success, and retries are
# automatic on transient failures.
summary: str = await ctx.step.run(
"openai-summary", call_openai_summary, format_thread(thread),
)
return {"summary": summary}
Dashboard में, यह run function का step trace दिखाता है (load-thread उसके बाद openai-summary) हर step के inputs और outputs के साथ। अगर OpenAI ने एक 429 (rate limited) return किया, Inngest openai-summary को backoff के साथ automatically retry करता है: वही memoization semantics जो Concept 7 की हैं, इसलिए retries पहले वाले load-thread step को double-bill नहीं करतीं। जो आपको नहीं मिलता (TypeScript के step.ai.wrap की तुलना में): automatic LLM-specific telemetry जैसे token counts, model name, और dashboard के AI view में broken out provider-specific traces. ज़्यादातर Python production workloads के लिए, standard step trace plus आपकी अपनी OpenAI client telemetry (मसलन, OpenAI Agents SDK का tracing) इस gap को cover करती है।
क्योंकि step.run हर step के inputs और outputs Inngest के observability store में record करता है, जो content आप एक step से pass करते हैं वह store होता है और dashboard में दिखता है। अगर आपके prompt में PII (names, emails, addresses), secrets (API keys, internal tokens), contractual या financial data, या regulated content (HIPAA, GDPR-scoped data, PCI) है, तो raw content को step body में pass न करें। Redact करें, hash करें, summarize करें, या एक reference pass करें (एक customer_id और ticket_id, पूरा ticket text नहीं) और sensitive content को step body के अंदर अपने authoritative store से फिर load करें, जहाँ retention और access controls configure करना आपके हाथ में है। यही discipline OpenAI Agents SDK के अपने tracing पर भी लागू होती है अगर आप उसे enable करते हैं। Step traces को किसी भी production log की तरह treat करें: default रूप से useful, policy से regulated.
step.ai.infer: serverless cost reduction के लिए एक niche tool (Python-supported). आप इसके लिए बहुत कम पहुँचेंगे; step.run इस course की हर AI call के लिए default है। step.ai.infer एक specific situation के लिए मौजूद है: अपनी function process से OpenAI को call करने के बजाय, आप Inngest के infrastructure से call करने को कहते हैं, ताकि जब request in flight हो तब आपकी function process deallocate हो सके। उन serverless platforms (Vercel, Cloudflare Workers, AWS Lambda) पर जो in-flight time के लिए bill करते हैं, यह wait के दौरान compute cost बचाता है। Long-running inferences (Deep Research, बड़े embedding batches) के लिए बचत real है। Sub-second calls के लिए, यह बिना ज़्यादा फ़ायदे के latency add करता है।
अगर आप कभी इसके लिए पहुँचते हैं, तो अपने installed version के लिए exact signature AI Inference docs से लें: यह experimental inngest.experimental.ai namespace में रहता है और इस course के build में exercise नहीं किया गया था।
Quick check. True या false. (a) Python में,
ctx.step.run("name", call_openai, ...)OpenAI call को durable, transient failures पर retried, और success पर memoized बनाता है। (b) Python में OpenAI Agents SDK के साथ Inngest use करने के लिएstep.ai.inferएक hard requirement है। (c) एक single OpenAI call के लिएstep.runकोstep.ai.inferसे बदलने से function चलाना हमेशा सस्ता हो जाएगा।
Answers: (a) True: यह recommended Python pattern है। OpenAI call step body के अंदर जाता है; Inngest पूरे step को काम की unit मानता है। (b) False: ज़्यादातर मामलों के लिए step.run काफ़ी है। step.ai.infer serverless compute cost के लिए एक optimization है, requirement नहीं। Worked example में OpenAI Agents SDK integration plain step.run use करता है। (c) False: step.ai.infer पैसे तभी बचाता है जब (i) आप एक ऐसे serverless platform पर हों जो in-flight time के लिए bill करता है AND (ii) call इतनी लंबी हो कि request-offload की बचत add की गई orchestration overhead पर हावी रहे। Always-on servers पर sub-second calls के लिए, plain step.run जीतता है।Try with AI
With my AI coding assistant: take a customer-support agent
invocation and produce TWO versions of the Inngest function that
calls it:
Version A: Wrap the Runner.run call in step.run (the recommended
Python pattern: durable, retried on transient failures, memoized;
you get the standard step trace).
Version B: For comparison, write a SEPARATE small Inngest function
that calls a single OpenAI completion via step.ai.infer (the
Python-supported step.ai primitive that offloads inference to
Inngest's infrastructure to save serverless compute cost).
For each version, explain (a) what the dashboard trace shows for a
successful run, (b) what happens when the OpenAI call hits a 429
rate limit, and (c) on which kind of deployment (always-on server
vs serverless) Version B's offload saves real money.
Part 3: balance और recovery, production scale
Balance तीसरी layer है: यह worker को load के नीचे healthy रखता है, उसी तरह जैसे आपका शरीर खुद को steady रखता है जब आप उसे ज़ोर से push करते हैं। Concurrency worker को downstream systems पिघलाने से रोकता है। Throttling आपको rate-limit की दीवारों से बचाए रखता है। Priority और fairness एक बातूनी customer को सबको starve करने से रोकते हैं। Batching "आधी रात को 10,000 events" को "100 manageable function runs" में बदलता है। Replay "कल के bug ने हमें 200 failed interactions में डाला" को "हमने उसे fix किया; 200 conversations resume हो गईं" में बदलता है। Human-approval gates agent को तब तक suspend करते हैं जब तक एक human approve न करे। Part 3 के पाँच concepts वे production policies देते हैं जो एक working worker को ऐसा बनाते हैं जिसे आप paying customers के सामने रख सकते हैं।
Concept 11: Concurrency and throttling
Concurrency एक function के उन runs की maximum संख्या है जो एक साथ execute हो सकती हैं। Throttling उन runs की maximum संख्या है जो per unit time start हो सकती हैं। दोनों per function एक-एक line से configure होती हैं। दोनों सबसे common production gap हैं जब teams prototype से scale पर जाती हैं।
from datetime import timedelta
@inngest_client.create_function(
fn_id="customer-support-conversation",
trigger=inngest.TriggerEvent(event="customer/email.received"),
concurrency=[inngest.Concurrency(limit=10)],
throttle=inngest.Throttle(limit=100, period=timedelta(minutes=1)),
)
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
...
concurrency=10 कहता है: किसी भी पल इनमें से ज़्यादा से ज़्यादा 10 functions चल रही हैं। 11वाँ event queue में wait करता है जब तक 10 में से एक finish न हो। throttle=100/minute कहता है: per minute ज़्यादा से ज़्यादा 100 नए runs start होते हैं। 101वाँ event wait करता है भले concurrency headroom हो।
व्यवहार में दोनों क्यों matter करते हैं। Concurrency downstream systems की रक्षा करता है: अगर आपका customer-support Worker OpenAI और Postgres से बात करता है, तो 1,000 concurrent runs का मतलब 1,000 एक साथ OpenAI calls और 1,000 एक साथ Postgres connections है। आप अपना OpenAI rate limit खत्म कर देंगे, अपना connection pool खत्म कर देंगे, या दोनों। Throttle bursts से रक्षा करता है: अगर 500 customer emails ठीक 9:00am पर आते हैं, आप नहीं चाहते कि 500 functions एक ही second में start हों; throttle start rate को smooth करता है।
Per-key concurrency. एक single concurrency limit function पर globally लागू होती है। एक ज़्यादा दिलचस्प pattern per-key concurrency है: event की किसी property से limit करना।
@inngest_client.create_function(
fn_id="customer-support-conversation",
trigger=inngest.TriggerEvent(event="customer/email.received"),
concurrency=[
inngest.Concurrency(limit=10), # global cap
inngest.Concurrency(limit=2, key="event.data.customer_id"), # per-customer cap
],
)
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
...
यह कहता है: globally ज़्यादा से ज़्यादा 10 functions चल रही हैं, AND एक समय में per customer ज़्यादा से ज़्यादा 2. अगर एक single customer एक minute में 100 emails भेजता है, सिर्फ़ उनके 2 emails एक साथ process होते हैं; बाक़ी 98 पीछे queue होते हैं। इस बीच, दूसरे customers के emails normally बहते हैं; वे बातूनी customer से block नहीं होते। यह दो lines code में multi-tenant fairness है। Concept 12 इस pattern को आगे develop करता है।
Quick check. तीन claims, True या False. (a) अगर आप
concurrency=10set करते हैं और 1,000 events एक साथ आते हैं, उनमें से 990 drop हो जाते हैं। (b) Throttling और concurrency limits दोनों total throughput घटाते हैं। (c) Per-key concurrency को एक ऐसी key चाहिए जो event data से deterministic हो।
Answers: (a) False: events drop नहीं होते; वे queue होते हैं। Inngest की queue durable है; 990 events तब तक wait करते हैं जब तक concurrency slots खुलें। (b) False. Throttling start-rate cap करता है; concurrency in-flight runs cap करता है। न तो कोई काम drop करता है; दोनों यह shape करते हैं कि काम कब execute होता है। एक लंबे window पर throughput unchanged रहता है अगर आपका average load limits से नीचे है। एक peak पर throughput shaped होता है: bursts queue सोख लेती है। (c) True: key expression event data पर evaluate होता है; उसे same logical scope के लिए एक stable string पैदा करना होता है (customer_id ठीक है; current_timestamp नहीं)।Try with AI
With my AI coding assistant: design the concurrency and throttling
policy for the customer-support Worker. Constraints:
- OpenAI rate limit: 30 requests per minute, hard cap.
- Postgres connection pool: 20 max connections (the Worker takes 1 per run).
- Some customers send bursts of 30+ emails in a minute (an angry
customer); these shouldn't starve other customers.
- We expect ~1,000 emails per day, with peaks around 9am and 2pm.
Propose:
1. A global concurrency value
2. A per-customer concurrency value
3. A throttle (limit and period)
For each, explain what production failure it protects against and
what the cost is (in queue latency at peak).
Concept 12: Priority and fairness, multi-tenant scaling
Concurrency limits काम करती हैं। Per-key concurrency basic fairness add करती है। Production-grade multi-tenant systems को और चाहिए: priorities (Enterprise customers को same compute के लिए hobbyists के पीछे wait नहीं करना चाहिए) और fair-share scheduling (कोई एक tenant अपने concurrency cap के अंदर भी system को monopolize न कर सके)।
Priority. Inngest हर event पर एक priority expression evaluate करता है; ज़्यादा priority वाले runs कम priority वाले runs से आगे queue में कूद जाते हैं।
@inngest_client.create_function(
fn_id="customer-support-conversation",
trigger=inngest.TriggerEvent(event="customer/email.received"),
concurrency=[inngest.Concurrency(limit=10)],
priority=inngest.Priority(
# Enterprise tier = high priority; Pro = 0; Free = low priority
run="100 - (event.data.customer_tier_priority * 100)",
),
)
async def handle_email(ctx: inngest.Context) -> dict[str, str]:
...
जब concurrency queue में 50 runs wait कर रहे हों, Enterprise customers के runs पहले जाते हैं, फिर Pro, फिर Free. Same tier के अंदर, FIFO order लागू होता है। Priority concurrency या throttle limits को override नहीं करती; यह बस तय करती है कि waiting runs में से कौन सा अगला free slot पाता है। एक Enterprise customer को फिर भी एक slot खुलने का wait करना पड़ता है; उन्हें बस अगला मिलता है।
Fair-share scheduling. जब आपके पास सैकड़ों tenants same global concurrency pool के लिए compete कर रहे हों, FIFO plus priority काफ़ी नहीं। एक single tenant जो एक burst भेजता है फिर भी कई minutes के लिए ज़्यादातर slots घेर सकता है। Fair-share scheduling, जो concurrency पर key parameter के ज़रिए एक सोच-समझकर रखे sizing से implement होती है, हर tenant को एक guaranteed slice देती है:
concurrency=[
inngest.Concurrency(limit=50), # global pool
inngest.Concurrency(limit=3, key="event.data.tenant_id"), # max 3 per tenant
],
इसके साथ: 50 total slots, कोई tenant 3 से ज़्यादा नहीं लेता। अगर 20 tenants active हैं, तो ज़्यादा से ज़्यादा 60 slots माँगे जाते हैं पर सिर्फ़ 50 available हैं। Fair-share उन्हें rotate करता है, हर tenant को कुछ share मिलता है, कोई बाहर नहीं रहता।
Predict. आपके पास
concurrency=10और per-customerconcurrency=2वाली एक customer-support function है। आपके पास priority भी configured है: Enterprise = high, Free = low. 9:00am पर, queue में हैं: Customer A (Free) से 5 events, Customer B (Enterprise) से 5 events, और एक single नए Customer C (Free, जिसने अभी अपना पहला plan ख़रीदा) से 10 events. वे किस order में execute होते हैं? Confidence 1-5.
Answer: यह एक multi-level decision है। पहले, per-customer cap 2 का मतलब है कि एक समय में हर customer के ज़्यादा से ज़्यादा 2 events run होने के eligible हैं। तो candidates का pool है: A से 2, B से 2, C से 2: छह runs तुरंत eligible. दूसरा, priority तय करती है कि उन छह में से कौन से पहले slots भरते हैं: B के दो पहले चलते हैं (Enterprise), फिर A के दो और C के दो (Free, FIFO). तो t=0 पर: B के 2 चलते हैं, फिर A के 2 start होते हैं, फिर C के 2 start होते हैं। Total: 6 active. जैसे-जैसे हर एक finish होता है, उसके customer का अगला queued event eligible बनता है और अगला slot priority से भरता है। यह वह तरह की policy है जो Inngest में एक feature है और आपके अपने code में एक हज़ार-line scheduler.Try with AI
With my AI coding assistant: extend the customer-support Worker
configuration with a priority and fair-share scheme. Requirements:
1. Three customer tiers: Enterprise, Pro, Free.
2. Enterprise customers should never wait more than 5 seconds at
peak load.
3. Free tier customers should get fair access: no Free customer
should be starved for more than 60 seconds, even when the
global queue is full.
4. A single noisy customer (regardless of tier) should not occupy
more than 3 slots.
Write the concurrency + priority configuration. For each line of
config, explain which requirement it satisfies.
Concept 13: Batching, cost-effective bulk processing
कुछ काम naturally batched होता है। आप 10,000 customer conversations में से हर एक को independently summarize नहीं करते; आप LLM को एक बार में 50 के batch के साथ call करते हैं। आप 10,000 audit rows एक-एक करके नहीं लिखते; आप उन्हें COPY करते हैं। Inngest का batch trigger आपको events accumulate करने और एक single function को batch के साथ input के रूप में invoke करने देता है।
@inngest_client.create_function(
fn_id="batch-embed-tickets",
trigger=inngest.TriggerEvent(event="ticket/resolved"),
batch_events=inngest.Batch(
max_size=50, # invoke when 50 events accumulated, OR
timeout=timedelta(seconds=30), # invoke when 30 seconds pass, whichever first
),
)
async def batch_embed_resolved_tickets(ctx: inngest.Context) -> dict[str, int]:
# ctx.events (plural) instead of ctx.event
ticket_ids = [e.data["ticket_id"] for e in ctx.events]
tickets = await ctx.step.run(
"load-tickets", load_tickets_by_ids, ticket_ids,
)
# One embedding call for 50 tickets, not 50 calls for 1 ticket each
embeddings = await ctx.step.run(
"embed-batch", embed_texts_batch,
[t["text"] for t in tickets],
)
await ctx.step.run(
"store-embeddings", store_embeddings_batch,
ticket_ids, embeddings,
)
return {"batched": len(ctx.events)}
क्या बदलता है: ctx.events एक list है, एक single event नहीं। Function per event एक बार के बजाय per batch एक बार चलती है। OpenAI embedding API को 50-text batch के साथ call किया जाता है, 50 single-text calls के बजाय, जो ज़बरदस्त रूप से सस्ता है (आप per token pay करते हैं, पर per-request overhead चला जाता है) और तेज़ है (50 के बजाय एक API round-trip)।
Batching तब सही tool है जब काम naturally bulkable हो (embeddings, bulk DB writes, bulk emails) और आप काम होने से पहले अपने timeout जितनी latency tolerate कर सकें। यह तब ग़लत tool है जब हर event को एक interactive response चाहिए या जब events के पार ordering unpredictable तरीक़ों से matter करती हो।
Quick check. True या false. (a) Batched functions को फिर भी retries और memoization मिलती है; पूरा batch durably memoized होता है। (b) अगर batch timeout सिर्फ़ 3 events accumulate होने पर खत्म हो जाता है, function तब तक नहीं चलेगी जब तक अगले 47 न आएँ। (c) आप
batch_eventsकोconcurrencyके साथ combine करके cap कर सकते हैं कि कितने batches parallel चलें।
Answers: (a) True: batch ही काम की unit है; retries पूरे batch को उसके सारे events scope में रखते हुए replay करती हैं। (b) False: यही तो timeout का पूरा point है। 30 seconds बाद function जो भी accumulate हुआ है उसके साथ चलती है, भले वह 1 event हो। (c) True: यह production pattern है। Batch plus concurrency मिलकर आपके downstream load को अच्छी तरह cap करते हैं।Try with AI
With my AI coding assistant: write a batched Inngest function that
embeds resolved support tickets, converting a per-ticket event
handler into one batched call.
Triggers: 'ticket/resolved' event, batched at 50 events or 30 seconds.
The function should:
1. Load the ticket bodies in one query
2. Call OpenAI embeddings API with a 50-text batch (faster + cheaper)
3. Store the embeddings
4. Emit a 'ticket/embedded' event per ticket for downstream consumers
Use grep_docs to find the OpenAI batch-embedding pattern.
Concept 14: Replay and bulk cancellation, production recovery
कभी सब कुछ एक साथ ग़लत हो जाता है। आपने एक bug ship किया; पिछले छह घंटों में एक हज़ार runs fail हुईं। या आपका downstream API 30 minutes के लिए down था; उस window में जिसने भी उसे call करने की कोशिश की वह मर गया। या आपने एक logic error खोजी और उसे fix करने के बाद एक दिन का काम फिर से करना चाहते हैं।
पहले, वह distinction जो सबको परेशान करता है। Inngest आपको एक failed step को फिर चलाने के दो तरीके देता है, और वे अलग व्यवहार करते हैं:
- Automatic retry (same run के अंदर)। जब एक step throw करता है, Inngest function को backoff के साथ retry करता है, ऊपर से re-enter करता हुआ। Completed steps memo से return होते हैं और re-execute नहीं होते; सिर्फ़ failing step फिर चलता है। यह memo-preserving resume है, वही जो आपने Quick Win में देखा, और वही जो "step 3 पर खर्च हुए $0.20 फिर से खर्च नहीं होते" property को सच बनाता है। यह automatic है और original run के अंदर होता है।
- Replay / Rerun (dashboard button, कई runs के पार)। यह आपके current deployed code के साथ ऊपर से एक बिल्कुल नया run शुरू करता है, हर step शुरू से re-execute होता हुआ (एक rerun को एक नया run id मिलता है और पहला step फिर चलता है, पुराने का resume नहीं)। तो practice में पुराने run का memo यहाँ आपको नहीं बचाता। यह incident recovery के लिए है, completed काम skip करने के लिए नहीं।
इन्हें अलग रखना ही पूरा concept है। Memo payoff automatic retry में रहता है; Replay एक fresh start है।
दो विपरीत recovery primitives. Replay कहता है "यह काम fail हुआ, मैं चाहता हूँ यह fixed code पर फिर चले।" Bulk cancellation कहता है "यह काम queue हुआ था पर अब मैं नहीं चाहता कि यह हो।" Same dashboard surface, विपरीत intent. ज़्यादातर teams को real traffic चलाने के पहले तीन महीनों के अंदर दोनों चाहिए होते हैं।
Replay recovery primitive है। Failed runs अपने पूरे step history, input event, और failed step के exception के साथ persist रहते हैं। Dashboard से आप Functions view खोलते हैं, उस function पर filter करते हैं जिसमें failed runs हैं, एक time window और एक failure pattern चुनते हैं (कोई specific error message या बस "all failures"), और Replay click करते हैं। Inngest हर एक को जो भी code अभी deployed है उस पर ऊपर से एक fresh run के रूप में schedule करता है।
Replay के बारे में समझने लायक तीन चीज़ें।
- Replay आपके current deployed code का use करता है। अगर आपने runs के fail होने और replay करने के बीच एक fix deploy किया, तो replayed runs new code use करते हैं। यही पूरा point है: उन runs की आबादी लें जो एक bug पर मरीं, fix ship करें, और सबको hands-off फिर चलाएँ।
- Replay हर step re-execute करता है; यह पुराने run का memo reuse नहीं करता। एक replayed run एक new run है, तो हर step fixed code पर शुरू से फिर चलता है। Cost के लिहाज़ से, per replayed run पूरी function के cost का plan करें, सिर्फ़ failed step का नहीं। जो चीज़ एक replay को एक दूसरा real-world side effect (एक duplicate refund, एक duplicate email) issue करने से रोकती है वह memo नहीं, उस side effect पर एक idempotency key है (Concept 4): आप request से एक stable key derive करते हैं (एक refund के लिए, कुछ ऐसा जैसे
(order_id, request_id)) और provider एक repeat को no-op मानता है। इस course का minimal worker उस key को संक्षिप्तता के लिए छोड़ देता है, उसका refund customer पर match करता है और unconditionally लिखता है, इसलिए एक production version कोई real पैसा हिलने से पहले एक add करेगा। Memo एक run के अंदर रक्षा करता है; idempotency key reruns के पार रक्षा करती है। - Replay opt-in है। Failed runs dashboard में तब तक बैठे रहते हैं जब तक आप उन पर action न लें। वे हमेशा retry नहीं होते; वे गायब नहीं होते। वे आपका इंतज़ार करते हैं।
Bulk cancellation इसका उलटा है। कभी आपके पास हज़ारों queued या sleeping runs होते हैं जो अब आप नहीं चाहते: एक campaign cancel हो गया, एक customer churn हो गया और आप अब उन्हें follow-up emails नहीं भेजना चाहते, एक feature rollback हो गई। Dashboard से आप एक function और एक time window या event filter चुनते हैं, और Cancel click करते हैं। Matching runs साफ़-सुथरे terminate होते हैं: उनके step.sleep और step.wait_for_event calls resume नहीं होते, queued runs start नहीं होते, in-flight runs cancellation के लिए check करते हैं और अगले step boundary पर exit करते हैं। Cancellation step boundary का सम्मान करती है; एक in-flight step.run terminate होने से पहले उस step को finish करता है जिसमें वह है, इसलिए आपको आधे-completed Stripe charges या torn DB writes नहीं मिलते।
Replay vs cancellation एक decision के रूप में। जब runs की एक आबादी के साथ कुछ ग़लत हो जाए, एक सवाल पूछें: क्या मैं चाहता हूँ कि यह काम succeed हो या मैं चाहता हूँ कि यह न हो? अगर काम succeed होना चाहिए (bug-fix recovery), replay. अगर काम नहीं होना चाहिए (cancelled campaign, churned customer, rolled-back feature), cancel. अगर आप unsure हैं (मसलन, failed runs में कुछ ऐसे हैं जिन्हें आप recover करना चाहते हैं और कुछ जिन्हें पहले fire ही नहीं होना चाहिए था), अपनी dashboard query को ज़्यादा संकरा filter करें ताकि हर subset को सही treatment मिले।
Practice में यह तीन patterns enable करता है:
- "हमने एक bug ship किया" recovery। Bad deploy के time window में failed runs खोजें, bug fix करें, fix ship करें, failures replay करें। Customer का experience: उनके email को एक घंटे reply नहीं मिला पर आख़िरकार मिल गया, बिना आपके कोई recovery code लिखे।
- "Campaign cancelled" rollback। एक welcome series जो 14 दिनों में तीन follow-up emails fire करती है; customer day 4 पर churn हो जाता है। आप day-7 और day-14 follow-ups नहीं भेजना चाहते। Matching
wait-for-eventऔरsleepruns bulk-cancel करें। - "Schema migration" replay। आपने बदला कि agent summaries कैसे format करता है; आप चाहते हैं कि कल के tickets new format के साथ फिर summarize हों। उन runs को खोजें (successful हों या नहीं) और replay करें; क्योंकि एक replay ऊपर से एक fresh run है, agent हर step new code पर फिर चलाता है, जो यहाँ ठीक वही है जो आप चाहते हैं। अपने side-effecting steps को idempotent रखें ताकि उन्हें फिर चलाने से double-charge या double-send न हो।
Dev-server MCP recovery को आपके coding agent से बाहर निकले बिना accessible बनाता है। Development के दौरान आप AI से कह सकते हैं कि एक failed run inspect करने के लिए get_run_status use करे, फिर fixed code पर event को फिर fire करके काम recover करे (उसे एक new event id दें, क्योंकि same id के साथ फिर fire करना Concept 4 की idempotency semantics से एक no-op में dedup हो जाता है)। Dashboard का Rerun button equivalent one-click path है। दोनों तरीक़ों से आपको current code पर एक fresh run मिलता है, memo-preserving resume नहीं।
Quick check. True या false. (a) एक dashboard Replay काम को new deployed code पर फिर चलाता है। (b) एक dashboard Replay original run के successful steps memo से return करता है और सिर्फ़ failed वाला फिर चलाता है। (c) एक failing run के अंदर automatic retry completed steps memo से return करता है और सिर्फ़ failing step फिर चलाता है। (d) एक in-flight function को bulk-cancel करना currently-executing
step.runको तेज़ी से terminate करने के लिए बीच में abort कर देगा।
Answers: (a) True: एक replay जो भी अभी deployed है उस पर ऊपर से एक fresh run है, यही वजह है कि यह bug-fix recovery का tool है। (b) False: यही trap है। एक replay एक new run है जो हर step ऊपर से re-execute करता है, तो पुराने run का memo carry over नहीं होता। जो एक replayed side effect को दो बार fire होने से रोकता है वह idempotency key है, memo नहीं। (c) True: यह memo-preserving path है, और वही जो आपने Quick Win में देखा। Completed step एक attempt पर बैठा रहता है जबकि failing step retry होता है। (d) False: cancellation step boundary का सम्मान करती है; current step.run run के terminate होने से पहले finish (या fail) होता है। यह torn writes रोकता है।Try with AI
Walk through a recovery scenario with my AI coding assistant:
Yesterday at 14:00 we deployed a change to the worker's agent step.
A bug in the new code made the agent step throw on every run.
From 14:00 to 18:00, 47 customer-support runs failed at that step.
At 18:30 we noticed, fixed the bug, and re-deployed.
Use the dev-server MCP's grep_docs to find Inngest's replay docs,
then:
1. Outline the exact dashboard steps to identify the 47 failed runs.
2. Explain what a dashboard Replay does for one of those runs: is it
a fresh run from the top on the fixed code, or a resume that
reuses the old run's memo? What does that mean for the cost of
replaying all 47?
3. Confirm whether the customers will see one reply or several if a
replayed run re-sends the email, and name the mechanism that
keeps it to one (hint: it is not memo).
4. Identify ONE scenario in this story where you'd prefer to
bulk-cancel instead of replay, and explain why.
Concept 15: step.wait_for_event के साथ HITL gates, runtime में Invariant 1
Agent Factory की Invariant 1 कहती है कि human ही principal है: authored intent, agent का autonomous judgment नहीं, वह है जिसका high-stakes decisions पर runtime को सम्मान करना है। यह वह एक जगह है जहाँ एक human mind loop में वापस क़दम रखता है। बाक़ी हर जगह nervous system अपने आप चलता है, reflex से; यहाँ यह रुकता है और एक person का इंतज़ार करता है। यह production में approval gates के रूप में दिखता है: agent analysis करता है, action draft करता है, पर action execute तब तक नहीं करता जब तक एक human approve न करे।
Inngest का step.wait_for_event (Concept 8) आज किसी भी platform पर इसका सबसे clean expression है। Agent decision के बिंदु तक चलता है, suspend होता है, और एक approval event का इंतज़ार करता है। Human review करता है (Slack में, एक admin UI में, email में) और approve या reject click करता है। Event fire होता है। Function human के verdict के साथ resume होती है और तदनुसार act करती है। यही spec-driven का runtime पर मतलब है: nervous system plan को enforce करता है, कि किस action को human चाहिए, किस order में, किस timeout के साथ। यह agent की reasoning को police नहीं करता; यह control करता है कि agent को क्या करने की अनुमति है।
@inngest_client.create_function(
fn_id="refund-with-hitl-gate",
trigger=inngest.TriggerEvent(event="customer/refund.investigated"),
concurrency=[inngest.Concurrency(limit=5)],
)
async def refund_with_gate(ctx: inngest.Context) -> dict[str, str]:
request_id = ctx.event.data["request_id"]
amount_cents = ctx.event.data["amount_cents"]
# Step 1: the agent's analysis (your worker, run durably)
analysis = await ctx.step.run(
"agent-investigates",
run_refund_investigation_agent,
request_id=request_id,
)
# Step 2: if the agent thinks refund is warranted AND amount > $100,
# gate behind human approval
needs_approval = analysis.recommends_refund and amount_cents >= 10_000
if needs_approval:
await ctx.step.run(
"notify-approver",
send_slack_approval_request,
request_id=request_id,
analysis=analysis,
amount_cents=amount_cents,
)
# === THE HITL GATE ===
approval = await ctx.step.wait_for_event(
"wait-for-human-approval",
event="refund/approval.decided",
timeout=timedelta(hours=24),
if_exp=f"async.data.request_id == '{request_id}'",
)
if approval is None:
# Timeout: no human responded in 24h. Escalate.
await ctx.step.run(
"escalate-timeout",
escalate_to_senior_reviewer,
request_id=request_id,
)
return {"status": "escalated_timeout"}
if not approval.data["approved"]:
await ctx.step.run(
"notify-rejected", notify_customer_rejected,
request_id=request_id,
)
return {"status": "rejected_by_human"}
# Either it was approved, or it didn't need approval
refund = await ctx.step.run(
"issue-refund", call_stripe_refund,
request_id=request_id, amount_cents=amount_cents,
)
await ctx.step.run(
"audit-approved-refund", audit_refund,
request_id=request_id, refund=refund,
approved_by="human" if needs_approval else "auto",
)
return {"status": "issued", "refund_id": refund["id"]}
Code में आप क्या देखते हैं: steps का एक sequence, बीच में एक wait_for_event. Runtime पर क्या हो रहा है:
- Agent चलता है (step 1, durably)।
- Function तय करती है कि gate लागू होता है या नहीं (in-code logic, side effects से मुक्त)।
- अगर gated: एक Slack notification fire होता है (step 2, durable)। Function suspend होती है। 24 घंटे तक zero compute consume होता है।
- Slack में एक human Approve या Reject click करता है। Admin backend
inngest_client.sendकोrefund/approval.decidedऔरrequest_idके साथ call करता है। - Inngest event को suspended function से match करता है (
if_expfilter सुनिश्चित करता है कि सिर्फ़ matching request IDs match हों)। Function अगली line पर resume होती है। - Function human के decision से या तो refund issue करती है या rejection notify करती है। दोनों paths decision और approver को audit करते हैं।
यही वह है जो Inngest को एक queue-plus-state-machine से qualitatively अलग बनाता है। HITL pattern एक primitive है। Function का code ऊपर से नीचे पढ़ता है, gate inline के साथ। कोई callback नहीं, कोई state restoration नहीं, कोई if state == waiting_for_approval: ... dispatching नहीं। Runtime suspend/resume mechanic handle करता है; आपका code policy express करता है।
एक later course Invariant 1 को architecturally develop करता है: authored intent, spec-driven workflows, manager-of-workers layer जो तय करती है कि कौन से gates कौन से actions पर लागू होते हैं। यह course आपको runtime primitive देता है। जब वह manager layer आती है, वह जो gate implement करती है वह बिल्कुल यही wait_for_event pattern होगा, बस fleet scale पर composed. Primitive को अभी जान लेने का मतलब है कि architectural pattern बाद में "एक समझदार composition" के रूप में पढ़ता है, "जादू" के रूप में नहीं।
यह वह keystone है जो आप Part 4 के Decision 5 में बनाते हैं: refund approval, durable बनाई हुई। यहाँ का concept shape है; worked example इसे एक real needs_approval tool से wire करता है और साबित करता है कि refund ठीक एक बार fire होता है।
Predict. आपके पास
timeout=timedelta(hours=24)से set एक HITL gate है। एक customer का refund request शुक्रवार 17:00 पर आता है। Weekend में कोई human online नहीं है। Gate का timeout शनिवार 17:00 पर fire होता है। आपका timeout handler एक blocked refund record करता है। Reviewer request सोमवार 9:00am पर पढ़ता है। Timeline से गुज़रें: weekend के दौरान कितने function runs active थे? Inngest ने कितने compute के लिए charge किया? Confidence 1-5.
Answer: weekend के दौरान zero active function runs. Function suspended थी: Inngest ने उसका state store किया, function को memory से बाहर page किया, और या तो event या timeout का इंतज़ार किया। Inngest suspended time के लिए bill नहीं करता। जब शनिवार 17:00 आया और timeout fire हुआ, function उन कुछ सौ milliseconds के लिए resume हुई जो blocked-refund audit row लिखने में लगे, फिर complete हो गई। यह तथ्य कि reviewer सोमवार तक नहीं देखता, worker की तरफ़ से कुछ cost नहीं देता। Inngest पर HITL workflows की economics उन polling-based queues से ज़बरदस्त रूप से अलग है जो आपको "क्या यह अब approve हुआ?" polling के हर second के लिए bill करती हैं।Try with AI
With my AI coding assistant: design a durable refund-approval gate.
Specification:
1. The agent investigates and decides a refund is warranted, but the
refund tool needs human approval before it runs.
2. The gate should:
- Notify the on-call reviewer with the agent's recommendation
- Wait up to 4 hours for the reviewer to approve or reject
- On approve: issue the refund.
- On reject: do not issue; record a blocked refund.
- On 4-hour timeout: do not issue; record a blocked refund.
3. Every branch (approve/reject/timeout) writes an audit row from a
small fixed set of action names, capturing what was decided.
Use the dev-server MCP's send_event to simulate each branch of
the reviewer's decision during testing.
Part 4: worked example, एक customer-support Production Worker
यहाँ आप build करते हैं। पहले worker (एक prompt), फिर उसके चारों ओर nervous system, per prompt एक layer. आप अपने coding agent को छोटे plain-English prompts में direct करते हैं और वह code लिखता है; नीचे दिखाए snippets हर layer की कुछ load-bearing lines हैं, files नहीं। पूरा implementation एक live dev server और एक real model के against end-to-end चलाया गया था, इसलिए जो shapes आप देखते हैं वही चलती हैं। अगर कोई signature unfamiliar लगे, तो आपका agent current docs check करता है।
Shape: सात prompts, उसी base पर जो आपने पहले set up की।
- D0 worker को ही बनाता है, standalone.
- D1 agent run को durable बनाता है।
- D2 एक event को उसे wake करने देता है।
- D3 एक daily cron add करता है जो fan out करता है।
- D4 flow control add करता है।
- D5 keystone है: refunds पर एक durable human-approval gate.
- D6 साबित करता है कि worker एक broken step survive करता है: completed काम दोहराए बिना retry, फिर recover.
शुरू करने से पहले। आपका environment Quick Win से पहले ही set up है: वही
ai-agent-nervous-systemfolder खोलें, Inngest औरneon-postgresSkills installed के साथ, आपकाOPENAI_API_KEYऔर आपका NeonDATABASE_URL.envमें, आपकेcustomersऔरaudit_logtables provisioned, और तीनों MCP servers (Neon, Context7,inngest-dev) wired. बस दो reminders:
- Dev server चल रहा है। अगर आपने इसे बंद कर दिया तो फिर start करें: इसके अपने terminal में
npx inngest-cli@latest dev. Dashboardhttp://127.0.0.1:8288पर है। (जब आप बाद में Inngest Cloud पर deploy करते हैं, free Hobby tier बिना credit card के $0 है; इसके ceilings Part 5 में हैं।)- नीचे दिए MCP calls के लिए एक casing note. Dev-server tool names
snake_caseहैं (send_event,get_run_status,invoke_function), लेकिन उनके parameterscamelCaseहैं (get_run_statusrunIdलेता है,invoke_functionfunctionIdलेता है)। Python SDK पूरे मेंsnake_caseहै; सिर्फ़ MCP call parameterscamelCaseहैं।
Brief
आप एक छोटा customer-support worker बनाते हैं और उसे एक Production Worker nervous system देते हैं। Worker अपने sample customers को Neon customers table से पढ़ता है (id, email, tier), एक incoming email का एक गर्मजोशी भरा reply draft करता है, refund सिर्फ़ human approval के साथ issue कर सकता है, और एक छोटे fixed set से हर action के लिए Neon audit_log table में एक audit row लिखता है: message_received, message_sent, refund_issued, refund_blocked. फिर सात prompts उसके चारों ओर Inngest add करते हैं: एक event उसे wake करता है, agent call durably चलता है, एक daily cron हर eligible customer पर एक health check fan out करता है, flow control concurrency और throttle cap करता है, refund एक durable human gate पर pause होता है, और एक replay path failed runs recover करता है।
आगे आने वाले prompts के बारे में एक note. हर एक उसी तरह लिखा गया है जैसे आप उसे actually एक coding agent को कहेंगे: छोटा, plain, उस पर भरोसा करते हुए कि वह detail संभाल लेगा। वे cold paste करने पर काम करते हैं, और इससे भी बेहतर अगर आप पहले agent से orient कराएँ ("read the project and tell me what you see, then ask me anything unclear before you start") जैसे-जैसे files बढ़ती हैं। Prompts मंज़िल हैं; पहले orient करना on-ramp है।
D0: worker बनाओ, standalone
आप कहाँ हैं: base खुला है, dev server चल रहा है, और आपका Neon store provisioned है, पर अभी कोई worker मौजूद नहीं। यह Decision standalone worker बनाता है; अंत तक यह एक sample email पर चलता है और Neon में एक audit row लिखता है.
Base पहले से एक AGENTS.md ship करता है जो आपके agent ने खुलते ही पढ़ लिया, इसलिए वह project जानता है; यही वजह है कि ये prompts छोटे रहते हैं। उसमें का वह एक नियम जो अपने दिमाग़ में रखने लायक है वह पूरे course का architectural invariant है: worker का अपना code कभी inngest से import नहीं करता. Agent और उसके tools plain Python रहते हैं; nervous system उन्हें बाहर से wrap करता है। वह separation, agent और nervous system को अलग रखना, वह है जो आपको बाद में Inngest को Temporal या Restate से swap करने और worker को untouched छोड़ने देता है।
आपका Neon system of record Quick Win से पहले ही provisioned है: customers और audit_log tables मौजूद हैं, और DATABASE_URL आपके .env में है। तो worker उस database को शुरू से ही read और write करता है। अब worker बनाएँ। यह paste करें:
Build me a minimal customer-support agent with the OpenAI Agents SDK, running in a local sandbox. It reads the sample customers from my Neon
customerstable (each row has an id, email, and tier), drafts a warm reply to an incoming customer email, and can issue a refund, but the refund tool needs human approval before it runs. Write an audit row into my Neonaudit_logtable for every action, using a small fixed set of action names and theDATABASE_URLin.env. Seed thecustomerstable with five sample rows first if it is empty. Keep it small; it exists to be wrapped, not shipped. Then run it on a sample email and show me the reply.
Creates: worker.py और db.py (एक flat project, कोई src/ nesting नहीं)। D1 तीसरी file के रूप में Inngest host add करता है। Agent Postgres तक DATABASE_URL से पहुँचता है, कभी Neon MCP server से नहीं, जो आपका build-time tool भर है.
Seed data इतना छोटा है कि page पर रखा जा सके, तीन tiers में पाँच sample customers, जिन्हें agent customers table में insert करता है:
[
{ "id": "cust_001", "email": "ada@example.com", "tier": "enterprise" },
{ "id": "cust_002", "email": "grace@example.com", "tier": "pro" },
{ "id": "cust_003", "email": "linus@example.com", "tier": "pro" },
{ "id": "cust_004", "email": "edsger@example.com", "tier": "standard" },
{ "id": "cust_005", "email": "alan@example.com", "tier": "standard" }
]
आपका agent दो छोटी Python files लिखता है। db.py Postgres access रखता है: DATABASE_URL पर एक छोटा pooled asyncpg connection, एक load_customers() read, और एक record() audit-write helper एक closed vocabulary के साथ, चार-item set से बाहर का कोई भी action name raise करता है, जो एक typo को एक silent bad row के बजाय एक loud error में बदल देता है। worker.py एक SandboxAgent है जिसमें दो tools हैं जो db.py में call करते हैं। इसकी सिर्फ़ एक line बाक़ी course के लिए load-bearing है, refund tool का decorator:
@function_tool(needs_approval=True)
def issue_refund(order_id: str, amount_cents: int, reason: str) -> str:
...
वह needs_approval=True agent को refund issue करने के बजाय pause करा देता है: run refund pending के साथ वापस आता है और एक human decide करता है। यह वह hook है जिस पर पूरा HITL keystone (D5) टँगा है। (यह floor हर refund को gate करता है, जो keystone को simple रखता है; एक production worker आम तौर पर सिर्फ़ एक threshold से ऊपर gate करता, Concept 15 का over-$100 pattern. दोनों तरह से wiring identical है।)
जो agent लिखता है उसमें confirm करने लायक एक structural note, क्योंकि D5 इस पर depend करता है: build_agent() और sandbox run_config() को अलग functions रखें। जब D5 एक paused run resume करता है तो वह agent को same tool shape में फिर बनाता है और same run_config() फिर pass करता है; saved state sandbox session carry नहीं करता, इसलिए resume को उसे फिर supply करना होता है। इन्हें अभी अलग factor कर दें और keystone बाद में एक छोटा step बन जाता है।
Done when: agent एक sample email पर चलता है और एक छोटा reply print करता है, और Neon audit_log table में एक new row है (इसे console में check करें, या अपने agent से Neon tools पर इसे वापस पढ़ने को कहें)। अगर email एक refund describe करता है, run refund tool पर issue करने के बजाय pause होता है; वह pause ही पूरा point है, और D5 उसे durable बनाता है।
इस Part के prompts एक frontier-class coding agent (Claude Sonnet या Opus, एक GPT-5-class model, या Gemini 2.5 Pro) मानकर चलते हैं। आप जो Inngest architecture सीख रहे हैं (events, steps, memoization, flow control) वह SDK-level है और जो भी model आपके agent को drive करे उसके साथ टिकता है। लेकिन build experience strong instruction-following पर टिकता है, ख़ासकर D5 keystone. एक weaker model पर, एक prompt पर एक से ज़्यादा बार iterate करने और file names spell out करने की उम्मीद रखें। Architecture टूटी नहीं है; prompting को बस ज़्यादा scaffolding चाहिए।
D1: agent run को durable बनाओ
आप कहाँ हैं: एक worker जो सिर्फ़ तब चलता है जब आप उसे call करें, run के बीच एक crash पर सब कुछ खो देता हुआ। यह Decision agent call को step.run में wrap करता है; अंत तक एक completed run dashboard में agent step को memoized दिखाता है.
Nervous system यहाँ शुरू होता है: पूरे agent call को एक single step.run में wrap करें ताकि वह durable और memoized हो। यह paste करें:
Wrap the agent run in an Inngest durable function so it survives crashes and retries transient failures. The whole agent call goes inside a single
step.runso it is memoized. Run it in local dev mode against the Inngest dev server, with a FastAPI host. Confirm a completed run shows the agent step memoized in the dashboard.
Creates: inngest_app.py (dev mode में एक Inngest client, एक helper में agent call, और एक FastAPI host जिसे dev server discover करता है).
जो shape matter करती है वह एक step.run है जो agent call को wrap करता है:
async def handle_customer_email(ctx: inngest.Context) -> dict:
email_text = ctx.event.data["email_text"]
outcome = await ctx.step.run("run-agent", functools.partial(_run_agent, email_text))
return {"replied": outcome["status"] == "done"}
जो agent लिखता है उसमें confirm करने लायक दो idioms. Step handler अपना कोई argument नहीं लेता, इसलिए functools.partial email_text को पहले से bind करता है, यही तरीक़ा है जिससे आप किसी भी step में data pass करते हैं, और आप इसे यहाँ से हर step पर देखेंगे। और agent helper plain Runner.run use करता है, streamed runner नहीं: यह वह path है जिस पर human-approval keystone (D5) बना है, इसलिए इसे शुरू से use करना D5 को एक rewrite के बजाय एक छोटा step बनाता है। Client is_production=False से construct होता है (Quick Win का dev-mode flag)।
इसे दो processes के रूप में चलाएँ, function host और वह dev server जो उसे ढूँढता है:
uv run uvicorn inngest_app:app --port 8000 --reload --log-level info # terminal 1: function host (your model key is sourced here; --reload picks up the D6 break/fix edits)
npx inngest-cli@latest dev -u http://127.0.0.1:8000/api/inngest # terminal 2: dev server, auto-discovers the host
Done when: dashboard handle-customer-email list करता है और एक completed run run-agent step दिखाता है। (आप इसे D2 में एक event से ठीक से wake करते हैं; अभी के लिए, function का discoverable होना काफ़ी है।)
यह load-bearing move क्यों है। Agent call महँगा हिस्सा है: model tokens, कई seconds. step.run के अंदर उसका result memoized होता है, इसलिए जब कोई बाद का step fail होता है और function retry होती है, agent फिर नहीं चलता। वह एक wrapping ही एक ऐसे worker, जो हर retry पर double-pay और double-act करता है, और एक ऐसे worker, जो हर महँगा काम ठीक एक बार करता है, इन दोनों के बीच का फ़र्क़ है।
D2: इसे एक event पर trigger करो
आप कहाँ हैं: एक durable function जो पहले से customer/email.received से triggered है (D1 का decorator), पर कोई audit trail नहीं। यह Decision agent के हर तरफ़ एक audit row add करता है; अंत तक एक real event एक run drive करता है जिसमें दोनों rows लिखी होती हैं.
Agent से पहले एक audit step और बाद में एक add करें, फिर worker को हाथ से चलाने के बजाय एक real event से wake करें। यह paste करें:
Make the worker wake on a
customer/email.receivedevent instead of being run by hand. Add an ingress audit step before the agent and a reply audit step after it. Send a test event and show me the run completing with both audit rows.
Edits: inngest_app.py (function को agent के हर तरफ़ एक audit step मिलता है).
Shape agent step के चारों ओर दो और step.run calls है:
customer_id = ctx.event.data.get("customer_id") # bound from the event, alongside D1's email_text
await ctx.step.run("audit-received", functools.partial(
db.record, "message_received", customer_id=customer_id, detail=email_text[:80]))
outcome = await ctx.step.run("run-agent", functools.partial(_run_agent, email_text))
await ctx.step.run("audit-sent", functools.partial(
db.record, "message_sent", customer_id=customer_id, detail=(outcome["reply"] or "")[:80]))
हर row closed set से एक action name use करती है: अंदर message_received, बाहर message_sent, और db.record उसे DATABASE_URL पर Neon audit_log table में लिखता है। Test event को agent से dev-server MCP के send_event tool के साथ भेजें (name: "customer/email.received", एक data object जिसमें email_text और customer_id हो)। Dev server कोई भी event accept करता है, इसलिए locally test करने के लिए आप कोई webhook configure नहीं करते; production में आप अपने email provider को एक Inngest webhook URL पर point करेंगे जो उसके payload को इस event में reshape करता है, जो एक dashboard setting है, code नहीं।
Done when: run complete होती है, trace order में तीन steps दिखाता है (audit-received, run-agent, audit-sent), और Neon audit_log table में उस customer के लिए एक message_received और एक message_sent row है।
दो audit steps क्यों, एक नहीं। हर एक अपना step.run है, इसलिए हर एक independently memoized है। अगर reply step fail होता है और function retry होती है, ingress row दो बार नहीं लिखी जाती (memo hit) और agent दो बार नहीं चलता (वह भी memoized)। Audit trail retries के पार exactly-once रहता है, वह property जो D6 साबित करेगा।
D3: एक daily cron जो fan out करता है
आप कहाँ हैं: एक worker जिसे दुनिया एक बार में एक email जगाती है। यह Decision एक daily cron add करता है जो हर eligible customer पर एक event fan out करता है; अंत तक हर एक को अपना durable child run मिलता है.
Scheduled काम add करें: एक daily cron जो हर Pro और Enterprise customer पर एक health-check event fire करता है, हर event अपना durable run trigger करता हुआ। यह paste करें:
Add a daily cron that fans out one
customer/health_check.requestedevent per Pro and Enterprise customer, each one idempotency-keyed so a re-delivered cron run never double-fires. Each child event triggers its own durable run that writes one audit row. Invoke the cron manually and show me one child run per eligible customer.
Creates: एक cron parent जो fan out करता है और एक event consumer जो हर child handle करता है, दोनों host के साथ registered.
दो shapes इस Decision को लेकर चलती हैं। Trigger एक one-line cron decorator है, और fan-out N events हैं जिनमें से हर एक एक idempotency key carry करता है:
@inngest_client.create_function(fn_id="daily-health-check", trigger=inngest.TriggerCron(cron="0 9 * * *"))
async def daily_health_check(ctx: inngest.Context) -> dict:
# ... select Pro/Enterprise customers, then:
events = [
inngest.Event(
name="customer/health_check.requested",
data={"customer_id": c["id"]},
id=f"health-{c['id']}-{ctx.event.id}", # idempotency key per (customer, cron run)
)
for c in eligible
]
await ctx.step.send_event("fan-out-health-checks", events)
Idempotency key load-bearing detail है: id=f"health-{customer}-{cron_run}" का मतलब है कि अगर same cron run दो बार deliver होता है (एक redeploy, एक retry), duplicate event drop हो जाता है, इसलिए हर customer को रोज़ ठीक एक check मिलता है। Consumer एक साधारण event-triggered function है जो एक audit row लिखती है। Cron को agent से MCP के invoke_function tool के साथ invoke करें (कल 09:00 का इंतज़ार न करें)। एक dev quirk: dev server crons सिर्फ़ तब fire करता है जब वह चल रहा हो; production उन्हें Inngest के always-on infrastructure पर चलाता है।
Done when: parent seconds में complete होता है और dashboard हर eligible customer पर एक customer-health-check child run दिखाता है, जिसमें standard-tier customers ठीक से skip हुए होते हैं।
Fan-out क्यों, एक loop नहीं। Parent customers को खुद process नहीं करता; वह N events भेजता है और return करता है। हर child अपना run है, isolated, independently retryable, अपनी concurrency से capped. एक function के अंदर एक loop उन्हें couple कर देता: एक slow customer बाक़ी को रोक देता, और एक crash पूरा batch खो देता। Fan-out वह तरीक़ा है जिससे एक scheduled wake-up N independent durable runs बन जाता है।
D4: flow control
आप कहाँ हैं: एक worker जो हर email handle करता है पर एक burst के नीचे उन सबको एक साथ fire कर देता। यह Decision तीन flow-control policies add करता है; अंत तक एक बीस-event burst cap के नीचे queue होता है, बिना dropped या duplicated rows के.
जब 9am पर पाँच सौ emails आते हैं, worker को एक साथ पाँच सौ model calls fire नहीं करने चाहिए: वह rate limit उड़ा देता है और बातूनी customer के पीछे सबको starve कर देता है। एक global concurrency cap, एक per-customer cap, और एक throttle add करें। यह paste करें:
Add flow control to the email handler: a global concurrency cap, a per-customer concurrency key so one noisy customer can't starve the rest, and a throttle to protect the OpenAI rate limit. Fire a burst of twenty events across five customers and show me they queue under the cap and all complete with no dropped or duplicated audit rows.
Edits: inngest_app.py (email function पर तीन decorator arguments).
ये तीन arguments ही lesson हैं, पूरा D4 इन्हीं में रहता है:
concurrency=[
inngest.Concurrency(limit=10), # global cap
inngest.Concurrency(limit=2, key="event.data.customer_id"), # per-customer cap
],
throttle=inngest.Throttle(limit=100, period=datetime.timedelta(minutes=1)),
तीन knobs, तीन काम। Global limit=10 cap करता है कि एक साथ कितने runs execute हों, दो real ceilings की रक्षा करता हुआ: model का rate limit, और आपका Neon connection budget. आपके connections को दो चीज़ें bound करती हैं, और वे अलग scales पर काम करती हैं। एक single worker replica के अंदर, सारे runs एक asyncpg pool share करते हैं, इसलिए pool का max_size वह है जो connections को flat रखता है चाहे कितने भी runs active हों (एक host पर एक बीस-run burst भी मुट्ठी भर pooled connections पर चलता है)। Replicas के पार, वह local pool अब मदद नहीं करता, replica दो का अपना pool है, इसलिए concurrency cap वह है जो total runs को bound करता है, और इसलिए total connections को, fleet-wide: limit=10 पर दस replicas एक सौ runs और लगभग एक सौ connections हैं, जिन्हें आप Neon के budget के against size करते हैं (free tier कुछ सौ pooled allow करता है)। Pool और cap मिलकर रक्षा हैं: pool एक replica को bound करता है, cap fleet को bound करता है। किसी एक के बिना, unpooled, uncapped replicas के पार एक पाँच-सौ-email burst Neon के accept करने से कहीं ज़्यादा connections खोल देता है। Per-customer limit=2 जो event.data.customer_id पर keyed है का मतलब है कि एक customer का burst ज़्यादा से ज़्यादा दो slots घेरता है, इसलिए एक account से एक flood दूसरों को कभी starve नहीं करता। throttle cap करता है कि per minute कितने runs start हों, एक spike को एक steady rate में smooth करता हुआ। एक function ज़्यादा से ज़्यादा दो concurrency policies carry करती है; global-plus-per-key जोड़ी common shape है। Burst agent से fire करें: send_event के ज़रिए पाँच customers के पार बीस customer/email.received events.
Done when: burst cap के नीचे queue होता है (running count 10 पर या उससे नीचे रहता है, और per customer 2 पर या उससे नीचे), हर run complete होती है, और Neon audit_log table में ठीक बीस message_received और बीस message_sent rows हैं। कोई dropped runs नहीं, कोई duplicates नहीं, और burst के नीचे कोई Neon connection-limit errors नहीं, इस single host पर asyncpg pool connections को flat रखता है (burst चलते हुए भी आप सिर्फ़ मुट्ठी भर in use देखेंगे), और cap वह है जो एक बार आप scale out करें तो replicas के पार उन्हें flat रखेगा।
ये policy क्यों हैं, code क्यों नहीं। इनमें से कुछ भी आपकी function body में नहीं रहता; यह configuration है जिसे runtime enforce करता है। Caps के बिना, एक burst या तो एक downstream system पिघला देता है या एक tenant को worker monopolize करने देता है। वही fairness हाथ से लिखना एक queue plus एक scheduler plus एक rate limiter है, सैकड़ों lines. यहाँ यह तीन decorator arguments है।
D5: refunds पर एक durable human-approval gate (keystone)
आप कहाँ हैं: एक worker जिसका refund pause (D0 का needs_approval=True) ephemeral है, running process में जी रहा। यह Decision उस pause को durable बनाता है; अंत तक run zero compute पर suspend होती है, एक real approval event का इंतज़ार करती है, और refund ठीक एक बार issue करने के लिए resume होती है.
वह ephemeral pause ही gap है: एक crash, एक deploy, या एक reviewer जो दोपहर भर ले लेता है, और pending refund चला गया। यह पूरे course का keystone है: pause को durable बनाओ, ताकि function zero compute पर suspend हो, जितनी देर लगे उतनी देर एक real approval event का इंतज़ार करे, फिर ठीक उसी agent run को resume करे। यह paste करें:
The refund approval is currently an in-process pause that a crash or a slow reviewer would lose. Make it durable: when the agent pauses on the refund, persist its serialized run state as the step's output, then suspend the whole function on
step.wait_for_eventwaiting for arefund/approval.decidedevent (give it a four-hour timeout and match it to this customer). When the decision arrives, rehydrate the state, apply approve or reject, and resume the agent so the refund fires exactly once. Drive a refund, show me the run suspended and waiting, send an approval, and show me exactly one refund audit row. Then do it again with a rejection and show me a blocked row and no refund.
Edits: inngest_app.py (agent helpers pause और resume करना सीखते हैं; email function को gate मिलता है).
यह Decision बाक़ियों से ज़्यादा code कमाता है, क्योंकि suspend-and-resume का नाच ही lesson है। जब agent pause होता है, वह अपना run state serialize करता है; जब decision आता है, आप उस state को rehydrate करते हैं, approve या reject apply करते हैं, और resume करते हैं:
async def _run_agent(email_text: str) -> dict:
agent = worker.build_agent()
result = await Runner.run(agent, email_text, run_config=worker.run_config())
if result.interruptions: # the refund tool paused for approval
return {"status": "needs_approval", "state": result.to_state().to_string()}
return {"status": "done", "reply": result.final_output}
async def _resume_agent(state_str: str, approved: bool, rejection_message: str | None) -> dict:
agent = worker.build_agent()
state = await RunState.from_string(agent, state_str)
for item in state.get_interruptions():
if approved:
state.approve(item)
else:
state.reject(item, rejection_message=rejection_message or "Refund denied.")
db.record("refund_blocked", detail=f"args={item.arguments}")
result = await Runner.run(agent, state, run_config=worker.run_config())
return {"status": "resumed", "reply": result.final_output}
Email function के अंदर, gate एक inline wait_for_event है जहाँ agent pause हुआ; decision एक resume step drive करता है:
decision = await ctx.step.wait_for_event(
"await-refund-approval",
event="refund/approval.decided",
timeout=datetime.timedelta(hours=4),
if_exp=f"async.data.customer_id == '{customer_id}'",
)
# (decision is None on timeout -> write a refund_blocked row and return)
resumed = await ctx.step.run("resume-agent", functools.partial(
_resume_agent, outcome["state"], bool(decision.data.get("approved")), decision.data.get("rejection_message")))
इसे ऊपर से नीचे पढ़ें: gate एक otherwise साधारण function में एक inline call है। कोई callback नहीं, कोई state-machine dispatch नहीं, invocations के पार कोई if status == waiting: branching नहीं। Runtime suspend और resume handle करता है; आपका code policy express करता है। चार details अपनी जगह कमाते हैं:
result.to_state().to_string()paused run को serialize करता है, और यहrun-agentstep का output बन जाता है, इसलिए यह durably stored होता है।to_state()synchronous है;to_string()वह string return करता है जिसे आप persist करते हैं।RunState.from_string(agent, s)awaited है (यह एक coroutine है) और उस stored string को सीधे लेता है। फिर आपstate.get_interruptions()परapproveयाrejectकरते हैं और resume करने के लिएRunner.run(agent, state, ...)call करते हैं। (एक resume approvals pending छोड़ सकता है, इसलिए real helper तब तक loop करता है जब तक कोई न बचे।)- Same
run_config()resume पर फिर pass होता है, और agent same tool shape में फिर बनता है। Serialized state sandbox session carry नहीं करता, इसलिए resume को उसे फिर supply करना होता है। यही वह एक detail है जो अगर छूट जाए तो resumed run fail कर देती है। (D0 ने ठीक इसी के लिएbuild_agentऔरrun_configको अलग factor किया।) if_expdecision को इस customer से match करता है (async.data.customer_id == '...'), इसलिए एक customer के लिए एक approval कभी किसी दूसरे customer का run resume नहीं करती।
इसे agent से drive करने के लिए: एक customer/email.received event भेजें जिसका email एक refund describe करता हो, run को await-refund-approval पर suspend होते देखें (dashboard उसे WAITING दिखाता है, run status RUNNING पर पर zero compute), फिर refund/approval.decided को {"approved": true, "customer_id": "cust_001"} के साथ send_event के ज़रिए भेजें। इसे {"approved": false} के साथ फिर करें।
Done when: approval पर, suspended run resume होती है और Neon audit_log table में ठीक एक refund_issued row है। Rejection पर, run resume होती है, audit में एक refund_blocked row है और कोई refund_issued नहीं, और agent का reply denial समझाता है।
यह keystone क्यों है। हर दूसरी layer (senses, reflexes, balance) worker को अपने आप correct या healthy रखती है। यह वह है जहाँ human mind एक high-stakes action पर loop में फिर प्रवेश करता है, durably, जितनी देर लगे, wait करते समय zero cost पर। इसका एक queue-plus-database-plus-poller version एक छोटा project है। यहाँ यह एक wait_for_event और एक resume है।
D6: साबित करो कि durability एक broken step survive करती है
आप कहाँ हैं: एक full worker जिसकी हर layer wrapped है। यह Decision वह property साबित करता है जिसने यह सब justify किया; अंत तक आपने एक broken run को अपने failing step को कई बार retry करते देखा है जबकि उसका completed audit step ठीक एक बार चलता है, फिर एक fresh run पर काम recover किया है.
साबित करने लायक आख़िरी property वही है जिसने यह सब justify किया, Concept 7 का memoization mechanic. आपने उसे वहाँ समझा; अब उसे अपने ही worker में साबित करें। यह paste करें:
Deliberately break the agent step so it fails, fire an event, and show me Inngest retrying it while the earlier audit step stays memoized, so the failing run writes its ingress audit row exactly once across all the agent retries. Then fix the step and recover the work, and show me the recovery completing.
Agent step को जानबूझकर तोड़ें (_run_agent के अंदर एक ValueError raise करें), अलग customers के लिए कुछ customer/email.received events fire करें, और हर run का trace पढ़ें। यह proof है, और यह हर failing run के अंदर है: audit-received एक completed attempt दिखाता है और अपनी row एक बार लिखता है; run-agent कई Attempts दिखाता है जैसे यह backoff के साथ retry होता है (Inngest कई attempts पर default करता है) और फिर fail होता है; audit-sent कभी नहीं चलता। Audit step एक attempt पर बैठा हुआ जबकि agent step चढ़ता हुआ, यही Concept 7 का memoization है, अब आपके अपने worker में दिखता हुआ: failing run सिर्फ़ एक message_received row लिखता है चाहे agent step कितनी भी बार retry हो।
फिर break को revert करें (अगर आपने host को --reload के साथ चलाया तो वह auto-reload होता है; नहीं तो उसे restart करें) और fixed code पर event को फिर fire करके काम recover करें (या, एक real bad-deploy batch के लिए, dashboard का Rerun button; दोनों ऊपर से एक fresh run शुरू करते हैं, Concept 14 में cover किया गया)। यहाँ वह हिस्सा है जो लोगों को हैरान करता है, और यह correct behavior है, bug नहीं: recovery एक बिल्कुल नया run है, इसलिए यह audit-received फिर चलाता है और अपनी अपनी message_received row लिखता है। एक break-then-recover के बाद, उस customer के पास वैध रूप से दो message_received rows हैं, एक failed run से, एक recovery से। Memoization एक within-run guarantee है; यह कभी दो अलग runs में नहीं फैलती।
Done when: failed run के trace में, audit-received एक attempt पर बैठा और एक row लिखी जबकि run-agent ने कई attempts जमा कीं और fail हुआ, वह एक-attempt-despite-N-retries ही memoization है, साबित हुई। फिर recovery run fixed code पर run-agent और audit-sent complete करती है। Neon audit_log को console में query करें (या अपने agent से Neon tools पर वापस पढ़ने को कहें): जिस customer को आपने broke-and-recovered किया उसके पास दो message_received rows होंगी (failed run plus recovery) और एक message_sent (सिर्फ़ recovery वहाँ तक पहुँची), जो बिल्कुल सही है। असली diagnostic per-run है, per-customer नहीं: एक single run का trace खोलें और confirm करें कि audit-received एक attempt दिखाता है। अगर एक run का trace ingress step को दो बार चलते दिखाता है, वह एक memoization bug है (आम तौर पर एक non-unique step name); दो rows जो दो अलग runs में फैली हैं वह नहीं।
यह bright line क्यों है। एक worker जो एक bad deploy पर customer काम खो देता है वह बस एक agent है जिसे आप call करते हैं। एक worker जो वही bad deploy लेता है, loudly fail होता है, broken step को उस काम को दोहराए बिना retry करता है जो वह पहले ही finish कर चुका था (agent step की कई attempts, पर ingress audit एक बार लिखी), और fix के बाद एक fresh run पर साफ़-सुथरा recover करता है, वह एक Production Worker है। Proof failed run का अपना trace है, एक ingress attempt against कई agent attempts, runs के पार एक row count नहीं।
इसी nervous system को minimal floor के बजाय अपने ही SandboxAgent worker पर point करें; wrapping identical है। और यह step.wait_for_event approval उस course के optional Decision 10 की hand-rolled run-state table की जगह लेता है: जो durable gate आपने अभी बनाया वही persistence layer है, इसलिए आप table delete कर सकते हैं।
अभी जो हुआ
आपने एक छोटा customer-support worker बनाया और उसे एक nervous system दिया, एक बार में एक layer. D0 के बाद worker के internals कभी नहीं बदले: वही SandboxAgent, वही दो tools, वही Neon Postgres audit trail. जो बदला वह उसके चारों ओर सब कुछ है। यह अब एक customer/email.received event पर और एक daily cron पर wake होता है जो हर eligible customer पर fan out करता है, durably चलता है (agent call step.run के अंदर), flow control का सम्मान करता है (global और per-customer concurrency, एक throttle), refunds को एक durable human approval पर gate करता है (step.wait_for_event), और failed runs replay करके एक bad deploy से recover करता है, audit trail के साथ जो दिखाता है कि किसी भी single run के अंदर हर step ठीक एक बार fire हुआ, चाहे वह run कितनी भी बार retry हुई हो।
Agent code वही है; उसकी पहुँच नहीं। आपने एक agent से शुरू किया जिसे आप operate करते हैं, उसे prompt करें, उसे देखें, फिर prompt करें। अब आपके पास एक worker है जो अपने आप operate करता है: दुनिया उसे wake करती है, उसके reflexes उसे failures के पार ले जाते हैं, वह load के नीचे अपना balance बनाए रखता है, और एक human सिर्फ़ वहाँ क़दम रखता है जहाँ stakes माँगते हैं। यही वह रेखा है जो opening ने खींची थी, एक agent जिसे आप operate करते हैं और एक FTE जो खुद operate करता है के बीच, और आपने उसके आर-पार अभी build किया।
बाक़ी जो concerns हैं वे scale पर observability, multi-worker coordination, और वह manager layer है जो तय करती है कि कौन से workers कौन सा traffic संभालते हैं। वह track में अगला course है। यह course production-ready execution की unit cover करता है; अगला उन units को एक workforce में compose करता है।
Part 5: यह course कहाँ ख़त्म होता है
एक Production Worker का cost shape
दो cost surfaces matter करते हैं: infrastructure cost (Inngest, और जो भी store तथा compute पर आप worker चलाते हैं) और inference cost (model tokens)। Infrastructure load बढ़ने पर लगभग flat रहता है; inference linearly scale करता है। नीचे की method ही सीखने लायक है; कोई भी dollar figure ship होते ही stale हो जाता है, इसलिए numbers को illustrative मानें और budget में कोई number डालने से पहले current pricing pages check करें।
Inngest pricing. Inngest per execution charge करता है: हर function run, plus हर step-level retry, एक execution गिना जाता है।
| Tier | Price | Executions / month | Concurrent steps | Notable |
|---|---|---|---|---|
| Hobby | $0 | 50,000 | 5 | 3 users, 50 realtime connections, no credit card |
| Pro | from $75 / month | 1,000,000 | 100+ | 1000+ realtime connections, 15+ users, 7-day trace retention |
| Enterprise | custom | custom | 500-50,000 | SAML / RBAC, 90-day trace retention, dedicated support |
Events pricing इसके ऊपर layer होती है: per day पहले 1-5M events included हैं; उससे ऊपर, overage लगभग $0.000050 per event से शुरू होता है और ज़्यादा volume पर घटता है। Pro 1M cap पार करने पर per additional 1M executions $50 add करता है।
यहाँ matter करने वाले Hobby-tier ceilings. 5-concurrent-step cap का मतलब है कि भले आप code में concurrency=Concurrency(limit=10) declare करें, platform का account-level cap आपको 5 पर रोक देता है। आपका code production के लिए correct है; free tier पर observed concurrency 5 है। step.sleep और step.sleep_until भी tier-bounded हैं: free Hobby plan पर सात दिन तक, paid plans पर एक साल तक (Inngest usage limits)।
Inference cost हावी रहता है। एक typical customer-support run per conversation कुछ हज़ार से दस हज़ार model tokens use करता है। अपनी per-token price को अपने tokens-per-email से, और उसे अपने emails-per-day से गुणा करें और आपके पास वह line है जो matter करती है; ज़्यादातर workers के लिए यह बाक़ी सब को बौना कर देती है। यही वह है जिसे आप optimize करते हैं। बाक़ी सब एक rounding error है। दो सबसे high-value levers: एक stable cached prompt prefix रखें (ताकि model दोहराए जाने वाले हिस्से को हर call पर full price के बजाय सस्ते cached rate पर bill करे), और easy turns को एक सस्ते model पर route करें।
तीन Inngest-specific cost levers जब आप optimization zone में हों:
- Pure functions को
step.runमें wrap न करें। अगर एक function के कोई side effects नहीं, उसे durability की ज़रूरत नहीं; उसे wrap करना बिना फ़ायदे के एक step-run charge add करता है।step.runको I/O और side effects के लिए बचाएँ। - Bulk paths के लिए
batch_eventsuse करें। एक 50-event batch एक function run है, 50 नहीं। step.sleepऔरstep.wait_for_eventसे सस्ते में suspend करें। Suspended functions suspension time के लिए bill नहीं करतीं। एक 3-दिन का delayed-followup एक 3-second वाले जितना ही cost करता है।
Scale पर shape: inference वह bill है जो traffic के साथ बढ़ता है; Inngest, आपका data store, और compute तुलनात्मक रूप से flat रहते हैं। यहाँ छपे किसी figure पर भरोसा करने के बजाय वही गुणा अपने real volume पर चलाएँ।
Swap guide: nervous system invariant है, platform नहीं
यह course हर layer पर Inngest का नाम लेता है। ऐसा इसलिए कि एक teaching example को concrete answers चाहिए, "जो भी orchestrator पसंद हो use करो" नहीं। लेकिन architecture किसी भी compliant alternative के साथ काम करता है। पाँच swaps जिनकी course का design explicitly उम्मीद करता है:
-
Trigger surface: Inngest events → Temporal signals, Restate handlers, AWS EventBridge + Lambda. हर platform के पास "यह code तब चलता है जब यह named चीज़ होती है" express करने का एक तरीक़ा है। Event names, payload shapes, और idempotency discipline सब transfer होते हैं। जो बदलता है: SDK का decorator syntax और dashboard.
-
Durable execution: Inngest
step.run→ Temporal activities, Restate handlers, custom Postgres-backed state machines. हर एक आपको "इस side-effecting call को memoize करो, transient failure पर retry करो, crash के बाद resume करो" semantics देता है। Temporal सबसे नज़दीकी analog है और पुराना, ज़्यादा enterprise-tested option. Restate सबसे नया है और इसका एक ज़्यादा functional-programming flavor है। Custom state machines वह हैं जो teams तब लिखती हैं जब वे एक managed platform adopt नहीं कर सकतीं; आम तौर पर 1,000-10,000 lines code जो Inngest जो free में देता है उसका ~70% फिर से बनाती हैं। -
HITL primitive:
step.wait_for_event→ Temporal काawait Workflow.execute_activity(approval_signal), Restate के awakeables, custom Redis/Postgres approval queues. Pattern वही है: function suspend होती है, एक external signal उसे resume करता है, audit decision capture करता है। Inngest का expression लिखने में सबसे clean है; Temporal का ज़्यादा verbose पर large scale पर battle-tested है। -
Cron scheduling: Inngest cron triggers → Kubernetes CronJobs + queue, GitHub Actions schedules, AWS EventBridge schedules. Cron triggers commodity हैं। Inngest का फ़ायदा cron होना नहीं है; यह है कि cron-triggered functions को event-triggered वालों जैसी ही durability/replay/flow-control मिलती है, automatically. दूसरे platforms आपसे वह खुद wire कराते हैं।
-
Flow control: Inngest concurrency + throttle → worker concurrency वाली Temporal task queues, Redis-backed rate limiters, AWS SQS message visibility timeouts. दूसरे platforms यह कर सकते हैं; Inngest इसे उस configuration density के साथ करता है जो हमने देखी (एक decorator argument)।
Production scale पर open companion के रूप में Dapr. नाम देने लायक एक ज़्यादा ambitious replacement: production scale पर Inngest के structural companion के रूप में Dapr Agents, उसी तरह जैसे OpenCode, Claude Code का है। Dapr Agents CNCF governance के तहत 23 मार्च 2026 को v1.0 GA तक पहुँचा (CNCF announcement, Dapr Agents core concepts)। DurableAgent production-ready class है; पुरानी Agent class deprecated है। Dapr तब चुनें जब Kubernetes-native deployment और multi-language SDKs, Inngest के local dev experience से ज़्यादा matter करें। Inngest बेहतर learning tool है (dashboard mental model को visible बनाता है); Dapr बेहतर scale tool है जब आप Inngest के tier ceilings पर पहुँच गए हों या K8s-native multi-language deployment चाहिए।
Inngest open source भी है (github.com/inngest/inngest; 1.0 release ने सितंबर 2024 में self-hosting support add किया) और Helm + KEDA के ज़रिए self-hostable है। Scale पर जो axes matter करते हैं वे governance, support, और maturity हैं: Inngest एक single vendor द्वारा governed है जिसकी self-hosting story नई है; Dapr CNCF-governed है जिसका production track record लंबा है।
| इस course का concept | Inngest primitive | Dapr production analogue | Teaching note |
|---|---|---|---|
| Scheduled काम | TriggerCron | Cron input binding / Dapr Scheduler | वही idea: समय Worker को wake करता है। Dapr आम तौर पर component configuration माँगता है। |
| Webhook/event ingress | Inngest webhook endpoint → event | HTTP endpoint, input bindings, या pub/sub ingress | Inngest ज़्यादा plumbing छुपाता है; Dapr infrastructure control देता है। |
| Internal events | inngest_client.send() | Dapr pub/sub | वही event-driven mental model; Dapr में broker pluggable है। |
| Fan-out | एक event कई functions trigger करता है | एक topic/event कई services consume करती हैं | वही architecture; Dapr broker/topic/subscriber composition use करता है। |
| Durable steps | step.run() + memoization | Dapr Workflows + activities | समान production purpose, अलग developer model. |
| बिना compute के waiting | step.sleep() | Durable workflow timers | दोनों wait करते समय एक process खुला रखने से बचते हैं। |
| Human approval gate | step.wait_for_event() | Workflow external events/signals, pub/sub, actors | Inngest expression simpler है; Dapr ज़्यादा composable है। |
| Retries | Function/step retries | Workflow/activity retries + resiliency policies | Dapr resiliency को workflow behavior के साथ-साथ एक runtime policy भी बनाता है। |
| Dead-letter / failed runs | Inngest dashboard failed runs + replay | Broker DLQ + workflow status/restart/manual tooling | Inngest यहाँ ज़्यादा turnkey है; Dapr ज़्यादा infrastructure-native है। |
| Flow control | Concurrency, throttling, priority, batching | Kubernetes scaling, app concurrency, broker controls, resiliency policies, bulk pub/sub | Dapr यह कर सकता है, पर यह एक decorator argument नहीं है। Inngest denser है। |
| Stateful coordination | wait_for_event, event keys, step state | Actors + state store + workflows | long-lived identity/stateful coordination के लिए Dapr Actors ज़्यादा मज़बूत हैं। |
| Agent runtime | आपका agent Inngest function के अंदर | DurableAgent / Dapr Agents v1.0 GA | Dapr Agents explicitly agent को workflow-backed और resumable बनाता है। |
यह table एक translation guide है, identical APIs का दावा नहीं। Inngest production pattern को एक compact developer experience के साथ सिखाता है: triggers, steps, waits, replay, और flow control एक product surface में। Dapr उसी production architecture को distributed-systems building blocks से implement करता है: bindings, pub/sub, workflows, actors, state, resiliency, और Kubernetes-native operations. Concepts सीधे transfer होते हैं; implementation style बदलता है। Dapr के bindings overview और Dapr Agents core concepts के against मई 2026 तक verified.
Production scale पर Dapr के लिए पहुँचने की तीन वजहें:
- CNCF-governed, charter से vendor-neutral: कोई single vendor platform या उस पर आपकी dependence को control नहीं करता।
- First-class Python के साथ polyglot। Dapr Agents Python-first है; वही agent code JavaScript, Go, .NET, Java, या PHP में लिखी services के साथ-साथ चल सकता है बिना किसी के एक दूसरा framework सीखे।
- Kubernetes पर design से horizontally scalable। अपने cluster में, एक managed offering (Diagrid Catalyst) में, या locally
dapr initके ज़रिए चलाएँ। Scaling story हर environment में वही architecture है।
Honest caveat: Dapr एक getting-started platform नहीं है। उसे production में चलाने का मतलब है Kubernetes, state store, pub/sub broker, placement service, observability, YAML components, sidecars. यह काफ़ी operational surface है जब आपका goal अभी भी patterns सीखना है, यही वजह है कि यह course Inngest पर शुरू होता है: एक command, और dashboard आ जाता है। Dapr के लिए तब पहुँचें जब patterns बैठ जाएँ और सवाल आपके control वाले infrastructure पर organizational scale पर चलाने की तरफ़ shift हो जाए।
पहले concepts Inngest और OpenAI Agents SDK पर सीखें: तेज़ feedback loop, minimal infrastructure, patterns पर focus. जब आप उस scale तक पहुँचें जहाँ Kubernetes governance, polyglot teams, या vendor-neutrality non-negotiable बन जाएँ, वही architectural patterns ऊपर दी translation table को key बनाकर Dapr पर lift होते हैं। Patterns transfer होते हैं; substrate बदलता है; जो आपने इस course में सीखा वह load-bearing knowledge बनी रहती है।
यह course (अभी) क्या cover नहीं करता
आपने जो worker बनाया वह thesis द्वारा रखी Seven Invariants में से चार satisfy करता है। ख़ासतौर पर: यह एक engine पर चलता है (Invariant 4, SandboxAgent), एक system of record के against (Invariant 5, audit trail), जिसमें दुनिया उसे call कर सकती है (Invariant 7, जो triggers आपने add किए), और एक gated decision पर human principal के साथ (Invariant 1, आंशिक: runtime mechanism यहाँ है, broader architectural pattern बाद में)। बाक़ी तीन Invariants, और वह broader architecture जो workers से एक workforce बनाता है, बाद के courses हैं। हर एक पर एक bullet:
- Invariant 2: हर human को एक delegate चाहिए। Edge पर एक personal agent जो आपका context रखता है, आपके judgment का प्रतिनिधित्व करता है, और workforce को काम broker करता है। Thesis इसके current realization के रूप में OpenClaw का नाम लेती है।
- Invariant 3: workforce को एक manager चाहिए। एक orchestrator जो काम assign करता है, budgets enforce करता है, execution audit करता है, hiring को एक callable capability के रूप में expose करता है। Thesis Paperclip का नाम लेती है।
- Invariant 6: workforce policy के तहत expandable है। एक meta-layer जहाँ एक authorized agent एक prompt generate करता है, एक runtime provision करता है, और एक new Worker register करता है, बिना किसी human को जगाए। Claude Managed Agents इसका एक realization है।
एक single worker जो events पर wake होता है, durably चलता है, और humans पर gate करता है, इस course द्वारा सिखाए architecture की सबसे छोटी unit है। अगला course उस worker को एक workforce में extend करता है: एक manager द्वारा coordinated कई workers, on demand expandable, triggers द्वारा woken, spec द्वारा governed. वही OpenAI Agents SDK foundation, वही audit habit, वही Inngest nervous system. Architecture invariant है।
इसमें actually अच्छा कैसे बनें
यह crash course पढ़ना आपको Production Workers बनाने में अच्छा नहीं बनाता। इसका use करना बनाता है। आप worker बनाकर शुरू करते हैं, उसे wrap करते समय friction महसूस करते हैं, और friction के हर टुकड़े को आपको सिखाने देते हैं कि वह किस concept से जुड़ा है।
इस course के लिए mapping:
- "मेरी function event आने पर fire क्यों नहीं होती?" → event name typo या namespace mismatch (Concept 3)। अपने
TriggerEventमें event name string कीinngest_client.sendवाले से byte-for-byte तुलना करें। - "मेरी function same logical event के लिए दो बार fire क्यों हुई?" → missing idempotency key (Concept 4)। Event में एक deterministic seed के साथ एक
id=add करें। - "मेरी function एक deploy के बाद काम 'खो' क्यों गई?" →
step.runके बाहर का code काम कर रहा था (Concept 7)। I/O और side effects को named steps में wrap करें। - "Customer दो बार charge क्यों हुआ?" → Stripe call
step.runके बाहर था, या step name unique नहीं था (Concepts 6 और 7)। Call को एक namedstep.runमें move करें; step name को function के अंदर globally unique बनाएँ। - "OpenAI 9am peak पर 429 errors क्यों return करता है?" → missing throttle (Concept 11)।
throttle=Throttle(limit=N, period=timedelta(minutes=1))add करें। - "एक customer के bursts दूसरे customers को starve क्यों करते हैं?" → missing per-key concurrency (Concept 12)। एक दूसरा
Concurrency(limit=2, key="event.data.customer_id")add करें। - "मेरा HITL gate weekend में silently fire क्यों हुआ?" → missing timeout handler जो audit में write करे (Concept 15)।
approval is Noneपर branch करें और audit row explicitly लिखें।
Architecture को एक बार में एक टुकड़ा बनाएँ। यही वजह है कि Part 4 सात prompts है, एक नहीं। Worker बनाएँ (D0)। Agent को step.run में wrap करें (D1) और देखें कि जब आप run के बीच जानबूझकर crash करते हैं तो क्या बदलता है। उसे एक event पर wake करें (D2)। Cron fan-out add करें (D3), फिर flow control (D4) तब जब आप actually एक rate limit hit कर चुके हों, फिर durable approval gate (D5) तब जब एक high-stakes action को actually एक human चाहिए हो। हर layer अपना learning है। एक बड़े rewrite में मिलाकर, वे एक दीवार हैं।
यह course जो discipline सिखाता है (events पर wake, durably run, humans पर gate, bugs पर replay) वह architectural invariant है। जो भी platform इसे implement करे, वह चार-property contract ही है जिसके लिए आप वास्तव में commit कर रहे हैं। यह Lindy bet है: आप उन हिस्सों पर build करते हैं जो टिके रहे, plain functions, SQL, एक typed language, एक event bus, इस season का wrapper नहीं। Product replaceable है; discipline नहीं।
Quick reference
Narrative course और during-build reference के बीच एक separator. नीचे के sections search करने के लिए हैं, ऊपर से नीचे पढ़ने के लिए नहीं। हर concept का one-line gist intro के collapsed cheat sheet में है; यह section during-build diagnostic, दो decision trees, और file layout है।
Decision tree: trigger surface चुनें
जब दुनिया में कोई नई चीज़ होती है, wake-up कहाँ से आता है?
- एक external system ने हमें एक HTTP request भेजी। → Webhook trigger. Source को Inngest dashboard में configure करें; transform के ज़रिए payload reshape करें; resulting event consume करें।
- एक schedule कहता है कि समय हो गया। → Cron trigger.
TriggerCron(cron="..."). UTC use करें; production crons तब भी fire होते हैं जब आपकी service deploy के बीच हो। - एक दूसरी Inngest function ने अपने run के दौरान एक event emit किया। → Event trigger.
TriggerEvent(event="ns/name.subtype"). Same name पर एक या कई functions subscribe करें। - एक interactive user एक immediate response का इंतज़ार कर रहा है। → यह कोई Inngest trigger नहीं है। Request/response को अपने normal web endpoint में रखें; अगर response में heavy काम शामिल है, request के अंदर से एक event fire करें और तुरंत return करें, Inngest को काम asynchronously संभालने देते हुए।
Decision tree: step primitive चुनें
मान लें एक function चल रही है और आपको कुछ करना है, आप कौन सा step.* call उठाते हैं?
- एक side-effecting call (API, DB, file write, agent invocation)। →
ctx.step.run("name", fn, ...). Default. Success पर memoized, transient failure पर retried. - एक serverless platform पर एक long-running OpenAI call जो in-flight time के लिए bill करता है। →
ctx.step.ai.infer(...). Inference को Inngest के infrastructure पर offload करता है ताकि आपकी function process deallocate हो सके। - Continue करने से पहले एक fixed duration wait करें। →
ctx.step.sleep("name", timedelta(...)). Durable; wait करते समय zero compute (free plan पर सात दिन तक, paid पर एक साल)। - एक external event का wait करें (human approval, sibling-function completion)। →
ctx.step.wait_for_event("name", event="...", timeout=..., if_exp=...). Durable; event आने पर resume होता है या timeout परNonereturn करता है। - Pure deterministic computation (एक string format करना, एक date compute करना)। → बस code लिखें। कोई
step.runज़रूरी नहीं; कोई charge नहीं।
File-location quick-ref
एक flat project, चार files, कोई src/ nesting नहीं:
ai-agent-nervous-system/
├── .claude/
│ └── skills/ # the four Inngest skills (installed in the Quick Win)
│ ├── inngest-setup/SKILL.md
│ ├── inngest-events/SKILL.md
│ ├── inngest-steps/SKILL.md
│ └── inngest-durable-functions/SKILL.md
├── db.py # Neon Postgres access: pooled asyncpg, load_customers, record (closed-vocabulary audit) (D0)
├── worker.py # the worker: SandboxAgent + 2 tools (D0)
├── inngest_app.py # the nervous system: Inngest functions + FastAPI host (D1-D5)
├── .env # OPENAI_API_KEY, DATABASE_URL, INNGEST_DEV=1
└── AGENTS.md # the base's rules file (read on open)
Customers और audit trail आपके Neon database में रहते हैं (Quick Win में provisioned, D0 में seeded), local files में नहीं। Worker (db.py, worker.py) D0 के बाद कभी नहीं बदलता। हर nervous-system layer (D1 से D5 तक) inngest_app.py edit करती है।
Diagnostic table, symptom → root cause → concept
| Symptom | First suspect | Concept to re-read |
|---|---|---|
| Expected event आने पर function कभी fire नहीं होती | Event name typo, namespace mismatch | C3 (webhooks), C5 (fan-out) |
| Same logical event के लिए function दो बार fire होती है | Missing idempotency key | C4 (idempotency) |
| Deploy के बाद function ने काम "खो" दिया | step.run के बाहर का code काम कर रहा है | C7 (memoization) |
| एक deploy के दौरान cron schedule fire नहीं हुआ | सिर्फ़ local dev server, production Inngest infra पर चलता है | C2 (cron) |
| एक refund के लिए customer दो बार charge हुआ | Stripe call step.run के बाहर, या step name unique नहीं | C6 (step.run), C7 (memoization) |
| 9am peak के दौरान OpenAI rate-limit errors | Missing throttle | C11 (concurrency + throttle) |
| एक customer के bursts दूसरे customers को starve करते हैं | Missing per-key concurrency | C12 (priority + fairness) |
| Function हमेशा के लिए suspended, कभी resume नहीं हुई | wait_for_event में event name भेजे जा रहे event से match नहीं करता | C8 (wait_for_event), C15 (HITL) |
| HITL timeout weekend में silently fire हुआ | Missing timeout handler जो audit में write करे | D5 (durable refund gate), C15 (HITL) |
| कल के failed runs dashboard से गायब हो गए | Runs manually replayed होने तक या retention window के बाद तक persist रहते हैं | C14 (replay) |
| Replay ने customers को फिर charge किया | Replay एक fresh run है जो हर step re-execute करता है; charge की कोई idempotency key नहीं थी | C4 (idempotency), C14 (replay एक fresh run है) |
| Function trace OpenAI prompt नहीं दिखाता | Step trace function inputs/outputs दिखाता है पर कोई LLM-specific prompt/token telemetry नहीं | C10 (Python step.run use करता है; LLM-specific telemetry को आपके अपने OpenAI client tracing की ज़रूरत है; step.ai.wrap के prompt-level traces सिर्फ़ TypeScript के लिए हैं) |
Appendix: optional lineage और एक Inngest cheat sheet
यह course अकेला खड़ा है: Part 4 worker को शुरू से बनाता है, इसलिए नीचे कुछ भी एक prerequisite नहीं। Context के लिए दो छोटे notes.
A.1: अगर आप Digital FTE course से आ रहे हैं
From Agent to Digital FTE course एक richer customer-support worker बनाता है: portable Skills, एक Postgres system of record, और एक custom MCP server. अगर आपने वह किया, तो आपके पास disk पर पहले से एक SandboxAgent worker है, और आप D0 का minimal floor skip कर सकते हैं: nervous system (D1 से आगे) को minimal floor के बजाय अपने ही worker पर point करें। Wrapping identical है। एक bonus: जो durable refund gate आप D5 में बनाते हैं (step.wait_for_event) वह उस course के optional Decision 10 की hand-rolled run-state table की जगह लेता है, इसलिए आप उसे delete कर सकते हैं। अगर आपने वह course नहीं किया, इस सब को नज़रअंदाज़ करें; D0 आपको वह सब देता है जो आपको चाहिए।
A.2: इस course में use होने वाले Inngest-specific essentials
अगर नीचे कुछ भी unfamiliar लगे, Part 4 में गोता लगाने से पहले corresponding doc page skim करें।
- Inngest client instantiation. Per Python project एक single
inngest.Inngest(app_id=...)instance, एक module से exported और जहाँ भी आप functions decorate करते हैं वहाँ imported. Python quick start। - Function decoration.
@inngest_client.create_function(fn_id=..., trigger=...). TriggerTriggerEvent,TriggerCron, या multi-trigger functions के लिए दोनों की एक list हो सकता है। ctx.step.run,ctx.step.sleep,ctx.step.wait_for_event,ctx.step.ai.infer. चार step primitives जो Python में आप जो लिखेंगे उसका 90% बनाते हैं। (TypeScript के पास एक पाँचवाँ है,step.ai.wrap, LLM-specific tracing के लिए; Python projects AI calls के लिएstep.runuse करते हैं।)inngest_client.send(events=[...]). अपने code में कहीं से भी events emit करें (functions के अंदर, agent tools के अंदर, CLI scripts से)। Idempotency के लिए एकid=use करें।- Dev server startup.
npx inngest-cli@latest dev.:8288पर चलता है। Dashboardhttp://127.0.0.1:8288पर. MCPhttp://127.0.0.1:8288/mcpपर. अगर:8288लिया हुआ है तो यह8289+use करता है; फिर host परINNGEST_BASE_URL=http://127.0.0.1:<port>set करें ताकि वह follow करे, सिर्फ़ MCP URL नहीं।
A.3: वे दो shifts जो actually मुश्किल हैं
इस course की सबसे मुश्किल चीज़ Inngest का syntax नहीं है। यह request से event तक का mental shift (Concept 1) और in-process execution से durable execution तक (Concept 6) है। वे दोनों बैठ जाएँ तो syntax mechanical है। अगर बाक़ी कुछ भी ज़रूरत से ज़्यादा मुश्किल लगे तो पहले Concepts 1 और 6 फिर पढ़ें।