From 7d6a0a3b5cfec2cfaa73b46ed760500e9fdb62e1 Mon Sep 17 00:00:00 2001
From: Ruben Manrique <rmanrique@render.com>
Date: Thu, 13 Aug 2026 11:20:18 -0500
Subject: [PATCH] update to new sdk

---
 README.md                              |  4 +-
 data-pipeline/README.md                | 32 +++++++++-------
 data-pipeline/main.py                  | 33 +++++++++-------
 etl-job/README.md                      | 26 ++++++-------
 etl-job/main.py                        | 22 +++++------
 file-analyzer/README.md                | 37 +++++++++---------
 file-analyzer/workflow-service/main.py | 26 +++++++------
 file-processing/README.md              | 18 ++++-----
 file-processing/main.py                | 50 +++++++++++++++----------
 hello-world/README.md                  | 24 +++++++-----
 hello-world/main.py                    | 13 ++++---
 openai-agent/README.md                 | 37 +++++++++---------
 openai-agent/main.py                   | 52 ++++++++++++--------------
 13 files changed, 202 insertions(+), 172 deletions(-)

diff --git a/README.md b/README.md
index 03b3bcc..649e4cd 100644
--- a/README.md
+++ b/README.md
@@ -16,7 +16,7 @@ Render Workflows support both Python and TypeScript. This repo contains Python e
 
 | Example | Use Case | Key Patterns | Extra Dependencies |
 |---------|----------|--------------|-------------------|
-| [**Hello World**](./hello-world/) | Learn workflow basics with simple number processing | Task definition, subtask calling with `await`, basic orchestration | None |
+| [**Hello World**](./hello-world/) | Learn workflow basics with simple number processing | Task definition, subtask stepping with `ctx.step`, basic orchestration | None |
 | [**ETL Job**](./etl-job/) | Process CSV data with validation and statistics | Subtasks, sequential processing, batch operations, data validation | None |
 | [**OpenAI Agent**](./openai-agent/) | AI customer support agent with tool calling | Tool calling, nested subtasks (3 levels deep), stateful workflows, dynamic orchestration | `openai` |
 | [**File Processing**](./file-processing/) | Batch process multiple file formats in parallel | Parallel execution with `asyncio.gather()`, multi-format handling, aggregation | None |
@@ -28,7 +28,7 @@ Render Workflows support both Python and TypeScript. This repo contains Python e
 The simplest possible workflow — learn the fundamentals through simple number processing.
 
 - Ultra-simple task definitions
-- Clear subtask calling examples
+- Clear `ctx.step` subtask examples
 - Subtasks in loops demonstration
 - Multi-step workflow orchestration
 - Heavily commented code explaining every pattern
diff --git a/data-pipeline/README.md b/data-pipeline/README.md
index 8c348bb..9a9244b 100644
--- a/data-pipeline/README.md
+++ b/data-pipeline/README.md
@@ -212,18 +212,20 @@ Using `asyncio.gather()` ensures all sources are fetched in parallel for maximum
 
 ### Stage 2: Transform
 
-**`transform_user_data`**: Combines data from all sources and enriches each user by calling subtasks:
+**`transform_user_data`**: Combines data from all sources and enriches each user by stepping subtasks:
 ```python
 for user in users:
     # SUBTASK CALL: Calculate metrics for this user
-    user_metrics = await calculate_user_metrics(user, transactions, engagement)
+    user_metrics = await ctx.step(
+        calculate_user_metrics, user, transactions, engagement
+    )
 
     # SUBTASK CALL: Enrich with geographic data
-    geo_data = await enrich_with_geo_data(user['email'])
+    geo_data = await ctx.step(enrich_with_geo_data, user['email'])
 
     enriched_users.append({**user_metrics, 'geo': geo_data})
 ```
-This demonstrates **sequential subtask calls per item** in a transformation loop.
+This demonstrates **sequential subtask steps per item** in a transformation loop.
 
 **`calculate_user_metrics`**: Calculates per-user metrics:
 - Total spent and refunded
@@ -283,9 +285,9 @@ This demonstrates **sequential subtask calls per item** in a transformation loop
 
 ```python
 # SUBTASK PATTERN: Launch multiple subtasks in parallel
-user_task = fetch_user_data(user_ids)
-transaction_task = fetch_transaction_data(user_ids)
-engagement_task = fetch_engagement_data(user_ids)
+user_task = ctx.step(fetch_user_data, user_ids)
+transaction_task = ctx.step(fetch_transaction_data, user_ids)
+engagement_task = ctx.step(fetch_engagement_data, user_ids)
 
 # SUBTASK CALLS: Wait for all three subtasks to complete
 user_data, transaction_data, engagement_data = await asyncio.gather(
@@ -304,15 +306,17 @@ Each user is enriched by calling multiple subtasks:
 ```python
 for user in users:
     # SUBTASK CALL: Calculate user-specific metrics
-    metrics = await calculate_user_metrics(user, transactions, engagement)
+    metrics = await ctx.step(
+        calculate_user_metrics, user, transactions, engagement
+    )
 
     # SUBTASK CALL: Enrich with geographic data
-    geo = await enrich_with_geo_data(user['email'])
+    geo = await ctx.step(enrich_with_geo_data, user['email'])
 
     enriched_users.append({**metrics, 'geo': geo})
 ```
 
-This shows **sequential subtask calls** for per-item enrichment.
+This shows **sequential subtask steps** for per-item enrichment.
 
 ### User Segmentation
 
@@ -327,7 +331,7 @@ Business logic classifies users into segments:
 **Add Real APIs**:
 ```python
 @app.task
-async def fetch_user_data_from_api(user_ids: list[str]) -> dict:
+async def fetch_user_data_from_api(ctx: TaskContext, user_ids: list[str]) -> dict:
     client = get_http_client()
     response = await client.post(
         "https://api.yourservice.com/users",
@@ -339,7 +343,7 @@ async def fetch_user_data_from_api(user_ids: list[str]) -> dict:
 **Add Database Integration**:
 ```python
 @app.task
-async def load_to_warehouse(insights: dict) -> dict:
+async def load_to_warehouse(ctx: TaskContext, insights: dict) -> dict:
     # Connect to data warehouse (Snowflake, BigQuery, etc.)
     # Insert aggregated insights
     # Return confirmation
@@ -349,7 +353,7 @@ async def load_to_warehouse(insights: dict) -> dict:
 **Add Caching**:
 ```python
 @app.task
-async def fetch_with_cache(source: str, key: str) -> dict:
+async def fetch_with_cache(ctx: TaskContext, source: str, key: str) -> dict:
     # Check Redis/Memcached
     # If miss, fetch from source and cache
     # Return data
@@ -359,7 +363,7 @@ async def fetch_with_cache(source: str, key: str) -> dict:
 **Add Notifications**:
 ```python
 @app.task
-async def send_pipeline_notification(result: dict) -> dict:
+async def send_pipeline_notification(ctx: TaskContext, result: dict) -> dict:
     # Send to Slack, email, etc.
     # Notify stakeholders of pipeline completion
     pass
diff --git a/data-pipeline/main.py b/data-pipeline/main.py
index 07ea6d8..90ad411 100644
--- a/data-pipeline/main.py
+++ b/data-pipeline/main.py
@@ -17,7 +17,7 @@
 import logging
 from datetime import datetime, timedelta
 
-from render_sdk import Retry, Workflows
+from render_sdk import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -55,7 +55,7 @@ def get_http_client():
 # ============================================================================
 
 @app.task
-async def fetch_user_data(user_ids: list[str]) -> dict:
+async def fetch_user_data(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Fetch user profile data from user service.
 
@@ -90,7 +90,7 @@ async def fetch_user_data(user_ids: list[str]) -> dict:
 
 
 @app.task
-async def fetch_transaction_data(user_ids: list[str], days: int = 30) -> dict:
+async def fetch_transaction_data(ctx: TaskContext, user_ids: list[str], days: int = 30) -> dict:
     """
     Fetch transaction history for users.
 
@@ -130,7 +130,7 @@ async def fetch_transaction_data(user_ids: list[str], days: int = 30) -> dict:
 
 
 @app.task
-async def fetch_engagement_data(user_ids: list[str]) -> dict:
+async def fetch_engagement_data(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Fetch user engagement metrics.
 
@@ -174,7 +174,7 @@ async def fetch_engagement_data(user_ids: list[str]) -> dict:
 # ============================================================================
 
 @app.task
-async def enrich_with_geo_data(user_email: str) -> dict:
+async def enrich_with_geo_data(ctx: TaskContext, user_email: str) -> dict:
     """
     Enrich user data with geographic information.
 
@@ -200,6 +200,7 @@ async def enrich_with_geo_data(user_email: str) -> dict:
 
 @app.task
 async def calculate_user_metrics(
+    ctx: TaskContext,
     user: dict,
     transactions: list[dict],
     engagement: dict
@@ -268,6 +269,7 @@ async def calculate_user_metrics(
 
 @app.task
 async def transform_user_data(
+    ctx: TaskContext,
     user_data: dict,
     transaction_data: dict,
     engagement_data: dict
@@ -303,11 +305,13 @@ async def transform_user_data(
         user_engagement = engagement_map.get(user['id'], {})
 
         # Calculate metrics for this user
-        user_metrics = await calculate_user_metrics(user, transactions, user_engagement)
+        user_metrics = await ctx.step(
+            calculate_user_metrics, user, transactions, user_engagement
+        )
 
         # Enrich with geo data
         user_email = user.get('email', f"{user['id']}@example.com")
-        geo_data = await enrich_with_geo_data(user_email)
+        geo_data = await ctx.step(enrich_with_geo_data, user_email)
         user_metrics['geo'] = geo_data
 
         enriched_users.append(user_metrics)
@@ -326,7 +330,7 @@ async def transform_user_data(
 # ============================================================================
 
 @app.task
-def aggregate_insights(enriched_data: dict) -> dict:
+def aggregate_insights(ctx: TaskContext, enriched_data: dict) -> dict:
     """
     Generate aggregate insights from enriched user data.
 
@@ -397,7 +401,7 @@ def aggregate_insights(enriched_data: dict) -> dict:
 # ============================================================================
 
 @app.task
-async def run_data_pipeline(user_ids: list[str]) -> dict:
+async def run_data_pipeline(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Execute the complete data pipeline.
 
@@ -426,9 +430,9 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
     try:
         # Stage 1: EXTRACT - Fetch from all sources in parallel
         logger.info("[PIPELINE] Stage 1/3: EXTRACT (parallel)")
-        user_task = fetch_user_data(user_ids)
-        transaction_task = fetch_transaction_data(user_ids)
-        engagement_task = fetch_engagement_data(user_ids)
+        user_task = ctx.step(fetch_user_data, user_ids)
+        transaction_task = ctx.step(fetch_transaction_data, user_ids)
+        engagement_task = ctx.step(fetch_engagement_data, user_ids)
 
         # Wait for all extractions to complete
         user_data, transaction_data, engagement_data = await asyncio.gather(
@@ -441,7 +445,8 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
 
         # Stage 2: TRANSFORM - Combine and enrich
         logger.info("[PIPELINE] Stage 2/3: TRANSFORM")
-        enriched_data = await transform_user_data(
+        enriched_data = await ctx.step(
+            transform_user_data,
             user_data,
             transaction_data,
             engagement_data
@@ -451,7 +456,7 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
 
         # Stage 3: LOAD - Generate insights
         logger.info("[PIPELINE] Stage 3/3: AGGREGATE")
-        insights = await aggregate_insights(enriched_data)
+        insights = await ctx.step(aggregate_insights, enriched_data)
 
         logger.info("[PIPELINE] Insights generated successfully")
 
diff --git a/etl-job/README.md b/etl-job/README.md
index f265807..9cd5261 100644
--- a/etl-job/README.md
+++ b/etl-job/README.md
@@ -13,7 +13,7 @@ Process customer signup data from CSV files with validation, cleaning, and stati
 
 ## Features
 
-- **Subtask Execution**: Demonstrates calling tasks from other tasks using `await`
+- **Subtask Execution**: Demonstrates running tasks from other tasks with `ctx.step`
 - **Extract**: Read data from CSV files (extensible to APIs, databases)
 - **Transform**: Validate records with comprehensive error tracking
 - **Load**: Compute statistics and prepare aggregated insights
@@ -163,23 +163,23 @@ This demonstrates how the pipeline handles data quality issues.
 - Validates age range (0-120)
 - Returns cleaned data with error tracking
 
-**`transform_batch`**: Processes all records by calling `validate_record` as a subtask for each one:
+**`transform_batch`**: Processes all records by running `validate_record` as a subtask for each one:
 ```python
 for record in records:
-    # Call validate_record as a subtask
-    validated = await validate_record(record)
+    # Run validate_record as a subtask on its own compute
+    validated = await ctx.step(validate_record, record)
 ```
-This demonstrates **calling subtasks in a loop** for batch processing.
+This demonstrates **stepping subtasks in a loop** for batch processing.
 
 **`compute_statistics`**: Aggregates valid records to produce:
 - Country distribution
 - Age statistics (min, max, average)
 - Data quality metrics
 
-**`run_etl_pipeline`**: Main orchestrator that calls three subtasks sequentially:
-1. `await extract_csv_data(source_file)` - Extract data
-2. `await transform_batch(raw_records)` - Validate records (which calls `validate_record` for each)
-3. `await compute_statistics(valid_records)` - Generate insights
+**`run_etl_pipeline`**: Main orchestrator that runs three subtasks sequentially:
+1. `await ctx.step(extract_csv_data, source_file)` - Extract data
+2. `await ctx.step(transform_batch, raw_records)` - Validate records (which steps `validate_record` for each)
+3. `await ctx.step(compute_statistics, valid_records)` - Generate insights
 
 This demonstrates **sequential subtask orchestration** for multi-stage pipelines.
 
@@ -188,7 +188,7 @@ This demonstrates **sequential subtask orchestration** for multi-stage pipelines
 **Add Database Loading**:
 ```python
 @app.task
-async def load_to_database(records: list[dict]) -> dict:
+async def load_to_database(ctx: TaskContext, records: list[dict]) -> dict:
     # Connect to database
     # Insert records
     # Return confirmation
@@ -198,7 +198,7 @@ async def load_to_database(records: list[dict]) -> dict:
 **Add API Data Source**:
 ```python
 @app.task
-async def extract_from_api(api_url: str) -> list[dict]:
+async def extract_from_api(ctx: TaskContext, api_url: str) -> list[dict]:
     # Fetch from REST API
     # Parse JSON response
     # Return records
@@ -210,9 +210,9 @@ async def extract_from_api(api_url: str) -> list[dict]:
 import asyncio
 
 @app.task
-async def transform_batch_parallel(records: list[dict]) -> dict:
+async def transform_batch_parallel(ctx: TaskContext, records: list[dict]) -> dict:
     # Validate all records in parallel
-    tasks = [validate_record(record) for record in records]
+    tasks = [ctx.step(validate_record, record) for record in records]
     results = await asyncio.gather(*tasks)
     # Aggregate results
     return results
diff --git a/etl-job/main.py b/etl-job/main.py
index 928754a..6985e8b 100644
--- a/etl-job/main.py
+++ b/etl-job/main.py
@@ -17,7 +17,7 @@
 from datetime import datetime
 from pathlib import Path
 
-from render_sdk import Retry, Workflows
+from render_sdk import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -38,7 +38,7 @@
 # ============================================================================
 
 @app.task
-def extract_csv_data(file_path: str) -> list[dict]:
+def extract_csv_data(ctx: TaskContext, file_path: str) -> list[dict]:
     """
     Extract data from a CSV file.
 
@@ -82,7 +82,7 @@ def extract_csv_data(file_path: str) -> list[dict]:
 # ============================================================================
 
 @app.task
-def validate_record(record: dict) -> dict:
+def validate_record(ctx: TaskContext, record: dict) -> dict:
     """
     Validate and clean a single data record.
 
@@ -143,7 +143,7 @@ def validate_record(record: dict) -> dict:
 
 
 @app.task
-async def transform_batch(records: list[dict]) -> dict:
+async def transform_batch(ctx: TaskContext, records: list[dict]) -> dict:
     """
     Transform a batch of records by validating each one.
 
@@ -165,8 +165,8 @@ async def transform_batch(records: list[dict]) -> dict:
     # KEY PATTERN: Calling subtasks in a loop
     for i, record in enumerate(records, 1):
         logger.info(f"[TRANSFORM] Processing record {i}/{len(records)}")
-        # SUBTASK CALL: Each record is validated by calling validate_record as a subtask
-        validated = await validate_record(record)
+        # SUBTASK CALL: ctx.step runs validate_record on its own compute for each record
+        validated = await ctx.step(validate_record, record)
 
         if validated['is_valid']:
             valid_records.append(validated)
@@ -193,7 +193,7 @@ async def transform_batch(records: list[dict]) -> dict:
 # ============================================================================
 
 @app.task
-def compute_statistics(valid_records: list[dict]) -> dict:
+def compute_statistics(ctx: TaskContext, valid_records: list[dict]) -> dict:
     """
     Compute statistical insights from validated records.
 
@@ -255,7 +255,7 @@ def compute_statistics(valid_records: list[dict]) -> dict:
 # ============================================================================
 
 @app.task
-async def run_etl_pipeline(source_file: str) -> dict:
+async def run_etl_pipeline(ctx: TaskContext, source_file: str) -> dict:
     """
     Complete ETL pipeline orchestrating extract, transform, and load operations.
 
@@ -282,20 +282,20 @@ async def run_etl_pipeline(source_file: str) -> dict:
         # Stage 1: Extract
         logger.info("[PIPELINE] Stage 1/3: EXTRACT")
         # SUBTASK CALL: Extract data from CSV
-        raw_records = await extract_csv_data(source_file)
+        raw_records = await ctx.step(extract_csv_data, source_file)
         logger.info(f"[PIPELINE] Extracted {len(raw_records)} records")
 
         # Stage 2: Transform
         logger.info("[PIPELINE] Stage 2/3: TRANSFORM")
         # SUBTASK CALL: Transform calls validate_record for each record
-        transform_result = await transform_batch(raw_records)
+        transform_result = await ctx.step(transform_batch, raw_records)
         logger.info(f"[PIPELINE] Transformation complete: "
                    f"{transform_result['success_rate']:.1%} success rate")
 
         # Stage 3: Load (compute statistics)
         logger.info("[PIPELINE] Stage 3/3: LOAD")
         # SUBTASK CALL: Compute final statistics
-        statistics = await compute_statistics(transform_result['valid_records'])
+        statistics = await ctx.step(compute_statistics, transform_result['valid_records'])
         logger.info("[PIPELINE] Statistics computed")
 
         # Build final result
diff --git a/file-analyzer/README.md b/file-analyzer/README.md
index a569bfa..858152d 100644
--- a/file-analyzer/README.md
+++ b/file-analyzer/README.md
@@ -141,25 +141,25 @@ file-analyzer/
 
 ### Tasks Defined
 
-**`parse_csv_data(file_content: str) -> dict`**
+**`parse_csv_data(ctx, file_content: str) -> dict`**
 - Parses CSV content into structured data
 - Returns rows, columns, and metadata
 
-**`calculate_statistics(data: dict) -> dict`** (Subtask)
+**`calculate_statistics(ctx, data: dict) -> dict`** (Subtask)
 - Calculates statistical metrics for numeric columns
 - Returns min, max, avg, sum for each numeric column
 
-**`identify_trends(data: dict) -> dict`** (Subtask)
+**`identify_trends(ctx, data: dict) -> dict`** (Subtask)
 - Identifies patterns in categorical data
 - Returns distribution analysis and top values
 
-**`generate_insights(stats: dict, trends: dict, metadata: dict) -> dict`** (Subtask)
+**`generate_insights(ctx, stats: dict, trends: dict, metadata: dict) -> dict`** (Subtask)
 - Generates final insights report
 - Combines statistics and trends into actionable findings
 
-**`analyze_file(file_content: str) -> dict`** (Main orchestrator)
+**`analyze_file(ctx, file_content: str) -> dict`** (Main orchestrator)
 - Coordinates the entire analysis pipeline
-- Calls parse → calculate → identify → generate as subtasks
+- Steps parse → calculate → identify → generate as subtasks
 
 ### Subtask Pattern
 
@@ -167,18 +167,18 @@ The main `analyze_file` task demonstrates subtask orchestration:
 
 ```python
 @app.task
-async def analyze_file(file_content: str) -> dict:
+async def analyze_file(ctx: TaskContext, file_content: str) -> dict:
     # SUBTASK CALL: Parse CSV data
-    parsed_data = await parse_csv_data(file_content)
+    parsed_data = await ctx.step(parse_csv_data, file_content)
 
     # SUBTASK CALL: Calculate statistics
-    stats = await calculate_statistics(parsed_data)
+    stats = await ctx.step(calculate_statistics, parsed_data)
 
     # SUBTASK CALL: Identify trends
-    trends = await identify_trends(parsed_data)
+    trends = await ctx.step(identify_trends, parsed_data)
 
     # SUBTASK CALL: Generate insights
-    insights = await generate_insights(stats, trends, parsed_data)
+    insights = await ctx.step(generate_insights, stats, trends, parsed_data)
 
     return {"statistics": stats, "trends": trends, "insights": insights}
 ```
@@ -522,12 +522,13 @@ print(result.results)  # Return value from task
 
 **Defining Tasks:**
 ```python
-from render_sdk import Workflows
+from render_sdk import TaskContext, Workflows
 
 app = Workflows()
 
+# Every task takes a TaskContext as its first parameter, followed by its inputs
 @app.task
-def my_task(param: str) -> dict:
+def my_task(ctx: TaskContext, param: str) -> dict:
     return {"result": param}
 
 app.start()
@@ -597,12 +598,14 @@ async def analyze_file(file: UploadFile):
 
 ```python
 @app.task
-async def analyze_file(file_content: str, webhook_url: str = None) -> dict:
+async def analyze_file(
+    ctx: TaskContext, file_content: str, webhook_url: str = None
+) -> dict:
     # ... perform analysis ...
 
     if webhook_url:
         # Notify completion
-        await send_webhook(webhook_url, results)
+        await ctx.step(send_webhook, webhook_url, results)
 
     return results
 ```
@@ -611,12 +614,12 @@ async def analyze_file(file_content: str, webhook_url: str = None) -> dict:
 
 ```python
 @app.task
-def parse_json_data(file_content: str) -> dict:
+def parse_json_data(ctx: TaskContext, file_content: str) -> dict:
     # Parse JSON files
     pass
 
 @app.task
-def parse_excel_data(file_content: bytes) -> dict:
+def parse_excel_data(ctx: TaskContext, file_content: bytes) -> dict:
     # Parse Excel files
     pass
 ```
diff --git a/file-analyzer/workflow-service/main.py b/file-analyzer/workflow-service/main.py
index ec55246..53873f5 100644
--- a/file-analyzer/workflow-service/main.py
+++ b/file-analyzer/workflow-service/main.py
@@ -17,7 +17,7 @@
 import io
 from datetime import datetime
 
-from render_sdk import Retry, Workflows
+from render_sdk import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -34,7 +34,7 @@
 
 
 @app.task
-def parse_csv_data(file_content: str) -> dict:
+def parse_csv_data(ctx: TaskContext, file_content: str) -> dict:
     """
     Parse CSV file content into structured data.
 
@@ -85,7 +85,7 @@ def parse_csv_data(file_content: str) -> dict:
 
 
 @app.task
-def calculate_statistics(data: dict) -> dict:
+def calculate_statistics(ctx: TaskContext, data: dict) -> dict:
     """
     Calculate statistical metrics from parsed data.
 
@@ -148,7 +148,7 @@ def calculate_statistics(data: dict) -> dict:
 
 
 @app.task
-def identify_trends(data: dict) -> dict:
+def identify_trends(ctx: TaskContext, data: dict) -> dict:
     """
     Identify trends and patterns in the data.
 
@@ -206,7 +206,9 @@ def identify_trends(data: dict) -> dict:
 
 
 @app.task
-async def generate_insights(stats: dict, trends: dict, metadata: dict) -> dict:
+async def generate_insights(
+    ctx: TaskContext, stats: dict, trends: dict, metadata: dict
+) -> dict:
     """
     Generate final insights report combining statistics and trends.
 
@@ -261,12 +263,12 @@ async def generate_insights(stats: dict, trends: dict, metadata: dict) -> dict:
 
 
 @app.task
-async def analyze_file(file_content: str) -> dict:
+async def analyze_file(ctx: TaskContext, file_content: str) -> dict:
     """
     Main orchestrator task for file analysis.
 
-    This task coordinates the entire analysis pipeline by calling
-    other tasks as SUBTASKS.
+    This task coordinates the entire analysis pipeline by running
+    other tasks as SUBTASKS via ctx.step.
 
     Pipeline:
     1. Parse CSV data
@@ -285,7 +287,7 @@ async def analyze_file(file_content: str) -> dict:
     # Stage 1: Parse CSV data
     logger.info("[ANALYZE_FILE] Stage 1: Parsing CSV data")
     # SUBTASK CALL: Parse the CSV content
-    parsed_data = await parse_csv_data(file_content)
+    parsed_data = await ctx.step(parse_csv_data, file_content)
 
     if not parsed_data["success"]:
         logger.error("[ANALYZE_FILE] Failed to parse CSV data")
@@ -300,17 +302,17 @@ async def analyze_file(file_content: str) -> dict:
     # Stage 2: Calculate statistics (SUBTASK)
     logger.info("[ANALYZE_FILE] Stage 2: Calculating statistics")
     # SUBTASK CALL: Calculate statistical metrics
-    stats = await calculate_statistics(parsed_data)
+    stats = await ctx.step(calculate_statistics, parsed_data)
 
     # Stage 3: Identify trends (SUBTASK)
     logger.info("[ANALYZE_FILE] Stage 3: Identifying trends")
     # SUBTASK CALL: Identify patterns and trends
-    trends = await identify_trends(parsed_data)
+    trends = await ctx.step(identify_trends, parsed_data)
 
     # Stage 4: Generate insights (SUBTASK)
     logger.info("[ANALYZE_FILE] Stage 4: Generating insights")
     # SUBTASK CALL: Generate final insights report
-    insights = await generate_insights(stats, trends, parsed_data)
+    insights = await ctx.step(generate_insights, stats, trends, parsed_data)
 
     logger.info("[ANALYZE_FILE] Analysis pipeline completed successfully")
 
diff --git a/file-processing/README.md b/file-processing/README.md
index 48674e1..8a176f6 100644
--- a/file-processing/README.md
+++ b/file-processing/README.md
@@ -261,9 +261,9 @@ The key to efficient batch processing is using `asyncio.gather()`:
 
 ```python
 @app.task
-async def process_file_batch(file_paths: list[str]) -> dict:
+async def process_file_batch(ctx: TaskContext, *file_paths: str) -> dict:
     # Launch all file processing tasks concurrently
-    tasks = [process_single_file(fp) for fp in file_paths]
+    tasks = [ctx.step(process_single_file, fp) for fp in file_paths]
     results = await asyncio.gather(*tasks)
 
     # Results from all files are ready
@@ -277,13 +277,13 @@ This processes all files simultaneously rather than sequentially, dramatically r
 **Add New File Format**:
 ```python
 @app.task
-def read_xml_file(file_path: str) -> dict:
+def read_xml_file(ctx: TaskContext, file_path: str) -> dict:
     # Parse XML file
     # Return structured data
     pass
 
 @app.task
-def analyze_xml_data(xml_result: dict) -> dict:
+def analyze_xml_data(ctx: TaskContext, xml_result: dict) -> dict:
     # Analyze XML content
     # Return insights
     pass
@@ -294,24 +294,24 @@ def analyze_xml_data(xml_result: dict) -> dict:
 **Add Cloud Storage Integration**:
 ```python
 @app.task
-async def download_from_s3(bucket: str, key: str) -> str:
+async def download_from_s3(ctx: TaskContext, bucket: str, key: str) -> str:
     # Download file from S3
     # Save to temp location
     # Return local path
     pass
 
 @app.task
-async def process_s3_batch(bucket: str, keys: list[str]) -> dict:
+async def process_s3_batch(ctx: TaskContext, bucket: str, keys: list[str]) -> dict:
     # Download files in parallel
-    paths = await asyncio.gather(*[download_from_s3(bucket, k) for k in keys])
+    paths = await asyncio.gather(*[ctx.step(download_from_s3, bucket, k) for k in keys])
     # Process files
-    return await process_file_batch(paths)
+    return await ctx.step(process_file_batch, *paths)
 ```
 
 **Add Database Export**:
 ```python
 @app.task
-async def export_to_database(report: dict) -> dict:
+async def export_to_database(ctx: TaskContext, report: dict) -> dict:
     # Connect to database
     # Insert report data
     # Return confirmation
diff --git a/file-processing/main.py b/file-processing/main.py
index 0a09a66..d890b7d 100644
--- a/file-processing/main.py
+++ b/file-processing/main.py
@@ -20,7 +20,7 @@
 from datetime import datetime
 from pathlib import Path
 
-from render_sdk import Retry, Workflows
+from render_sdk import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -41,7 +41,7 @@
 # ============================================================================
 
 @app.task
-def read_csv_file(file_path: str) -> dict:
+def read_csv_file(ctx: TaskContext, file_path: str) -> dict:
     """
     Read and parse a CSV file.
 
@@ -89,7 +89,7 @@ def read_csv_file(file_path: str) -> dict:
 
 
 @app.task
-def read_json_file(file_path: str) -> dict:
+def read_json_file(ctx: TaskContext, file_path: str) -> dict:
     """
     Read and parse a JSON file.
 
@@ -134,7 +134,7 @@ def read_json_file(file_path: str) -> dict:
 
 
 @app.task
-def read_text_file(file_path: str) -> dict:
+def read_text_file(ctx: TaskContext, file_path: str) -> dict:
     """
     Read and analyze a text file.
 
@@ -189,7 +189,7 @@ def read_text_file(file_path: str) -> dict:
 # ============================================================================
 
 @app.task
-def analyze_csv_data(csv_result: dict) -> dict:
+def analyze_csv_data(ctx: TaskContext, csv_result: dict) -> dict:
     """
     Analyze CSV data and extract insights.
 
@@ -247,7 +247,7 @@ def analyze_csv_data(csv_result: dict) -> dict:
 
 
 @app.task
-def analyze_json_structure(json_result: dict) -> dict:
+def analyze_json_structure(ctx: TaskContext, json_result: dict) -> dict:
     """
     Analyze JSON structure and extract metadata.
 
@@ -290,7 +290,7 @@ def count_keys(obj, depth=0):
 
 
 @app.task
-def analyze_text_content(text_result: dict) -> dict:
+def analyze_text_content(ctx: TaskContext, text_result: dict) -> dict:
     """
     Analyze text content for insights.
 
@@ -345,7 +345,7 @@ def analyze_text_content(text_result: dict) -> dict:
 # ============================================================================
 
 @app.task
-async def process_single_file(file_path: str) -> dict:
+async def process_single_file(ctx: TaskContext, file_path: str) -> dict:
     """
     Process a single file based on its extension.
 
@@ -362,22 +362,34 @@ async def process_single_file(file_path: str) -> dict:
     extension = path.suffix.lower()
 
     # Read file based on type
-    # SUBTASK PATTERN: Chain multiple subtask calls together
+    # SUBTASK PATTERN: Chain multiple ctx.step calls together
     if extension == '.csv':
         # SUBTASK CALL: Read CSV file
-        read_result = await read_csv_file(file_path)
+        read_result = await ctx.step(read_csv_file, file_path)
         # SUBTASK CALL: Analyze the CSV data (if read was successful)
-        analysis = await analyze_csv_data(read_result) if read_result.get("success") else {}
+        analysis = (
+            await ctx.step(analyze_csv_data, read_result)
+            if read_result.get("success")
+            else {}
+        )
     elif extension == '.json':
         # SUBTASK CALL: Read JSON file
-        read_result = await read_json_file(file_path)
+        read_result = await ctx.step(read_json_file, file_path)
         # SUBTASK CALL: Analyze JSON structure
-        analysis = await analyze_json_structure(read_result) if read_result.get("success") else {}
+        analysis = (
+            await ctx.step(analyze_json_structure, read_result)
+            if read_result.get("success")
+            else {}
+        )
     elif extension == '.txt':
         # SUBTASK CALL: Read text file
-        read_result = await read_text_file(file_path)
+        read_result = await ctx.step(read_text_file, file_path)
         # SUBTASK CALL: Analyze text content
-        analysis = await analyze_text_content(read_result) if read_result.get("success") else {}
+        analysis = (
+            await ctx.step(analyze_text_content, read_result)
+            if read_result.get("success")
+            else {}
+        )
     else:
         logger.warning(f"[PROCESS] Unsupported file type: {extension}")
         return {
@@ -398,7 +410,7 @@ async def process_single_file(file_path: str) -> dict:
 
 
 @app.task
-async def process_file_batch(*file_paths: str) -> dict:
+async def process_file_batch(ctx: TaskContext, *file_paths: str) -> dict:
     """
     Process multiple files in parallel.
 
@@ -418,9 +430,9 @@ async def process_file_batch(*file_paths: str) -> dict:
     logger.info("=" * 80)
 
     # Process all files in parallel
-    # SUBTASK PATTERN: Call multiple subtasks concurrently using asyncio.gather()
+    # SUBTASK PATTERN: Step multiple subtasks concurrently using asyncio.gather()
     logger.info("[BATCH] Launching parallel file processing tasks...")
-    tasks = [process_single_file(fp) for fp in file_paths_list]
+    tasks = [ctx.step(process_single_file, fp) for fp in file_paths_list]
     results = await asyncio.gather(*tasks)
 
     # Aggregate results
@@ -453,7 +465,7 @@ async def process_file_batch(*file_paths: str) -> dict:
 
 
 @app.task
-async def generate_consolidated_report(batch_result: dict) -> dict:
+async def generate_consolidated_report(ctx: TaskContext, batch_result: dict) -> dict:
     """
     Generate a consolidated report from batch processing results.
 
diff --git a/hello-world/README.md b/hello-world/README.md
index 47a0362..1c8e754 100644
--- a/hello-world/README.md
+++ b/hello-world/README.md
@@ -9,29 +9,32 @@ This hello-world example demonstrates three foundational workflow patterns:
 ## What You'll Learn
 
 - How to define tasks with `@app.task`
-- How to chain task runs using `await` and `asyncio.gather`
+- How to run one task from another with `ctx.step`
+- How to run task steps in parallel with `asyncio.gather`
 - How to customize retry behavior with `Retry`
 
 ## Example Tasks
 
-### `calculate_square(a: int) -> int`
+### `calculate_square(ctx, a: int) -> int`
 
-The smallest possible task: takes one integer and returns its square.
+The smallest possible task: takes one integer and returns its square. Like every
+task, it receives a `TaskContext` as its first parameter.
 
-### `sum_squares(a: int, b: int) -> int`
+### `sum_squares(ctx, a: int, b: int) -> int`
 
 Chains two runs of `calculate_square` and sums the results.
 
-It uses `asyncio.gather(...)` to chain the two runs in parallel:
+It uses `ctx.step(...)` to run each one on its own compute, and
+`asyncio.gather(...)` to run them in parallel:
 
 ```python
 result1, result2 = await asyncio.gather(
-    calculate_square(a),
-    calculate_square(b),
+    ctx.step(calculate_square, a),
+    ctx.step(calculate_square, b),
 )
 ```
 
-### `flip_coin() -> str`
+### `flip_coin(ctx) -> str`
 
 Simulates a coin flip:
 
@@ -92,7 +95,10 @@ Any function decorated with `@app.task` is registered when your service starts v
 
 ### Subtasks
 
-Inside an `async` task, calling `await other_task(...)` runs that task as a subtask.
+Every task receives a `TaskContext` as its first parameter. Inside an `async`
+task, `await ctx.step(other_task, ...)` runs `other_task` on its own compute and
+returns its result. The context parameter isn't part of the task's inputs, so
+`--input` only supplies the parameters that follow it.
 
 ### Retries
 
diff --git a/hello-world/main.py b/hello-world/main.py
index e088eb0..d2da17d 100644
--- a/hello-world/main.py
+++ b/hello-world/main.py
@@ -1,4 +1,4 @@
-from render_sdk import Workflows, Retry
+from render_sdk import TaskContext, Workflows, Retry
 import asyncio
 import random
 
@@ -6,15 +6,16 @@
 
 
 @app.task
-def calculate_square(a: int) -> int:
+def calculate_square(ctx: TaskContext, a: int) -> int:
     return a * a
 
 
 @app.task
-async def sum_squares(a: int, b: int) -> int:
+async def sum_squares(ctx: TaskContext, a: int, b: int) -> int:
+    # ctx.step runs a task on its own compute and returns its result
     result1, result2 = await asyncio.gather(
-        calculate_square(a),
-        calculate_square(b),
+        ctx.step(calculate_square, a),
+        ctx.step(calculate_square, b),
     )
     return result1 + result2
 
@@ -26,7 +27,7 @@ async def sum_squares(a: int, b: int) -> int:
         backoff_scaling=1.5,
     )
 )
-def flip_coin() -> str:
+def flip_coin(ctx: TaskContext) -> str:
     if random.random() < 0.5:
         raise Exception("Flipped tails! Retrying.")
     return "Flipped heads!"
diff --git a/openai-agent/README.md b/openai-agent/README.md
index b52f75c..701061f 100644
--- a/openai-agent/README.md
+++ b/openai-agent/README.md
@@ -274,34 +274,35 @@ Searches the knowledge base for information.
 **`execute_tool`**: Dynamically executes a tool as a subtask based on the agent's decision:
 ```python
 @app.task
-async def execute_tool(tool_name: str, arguments: dict) -> dict:
-    # Map tool names to tasks
-    tool_map = {
-        "get_order_status": get_order_status,
-        "process_refund": process_refund,
-        "search_knowledge_base": search_knowledge_base
-    }
-
-    # SUBTASK CALL: Execute the appropriate tool function
-    result = await tool_map[tool_name](**arguments)
-    return result
+async def execute_tool(ctx: TaskContext, tool_name: str, arguments: dict) -> dict:
+    # SUBTASK CALL: ctx.step runs the chosen tool on its own compute.
+    # Each tool takes different arguments.
+    if tool_name == "get_order_status":
+        return await ctx.step(get_order_status, arguments["order_id"])
+    if tool_name == "process_refund":
+        return await ctx.step(
+            process_refund, arguments["order_id"], arguments["reason"]
+        )
+    if tool_name == "search_knowledge_base":
+        return await ctx.step(search_knowledge_base, arguments["query"])
+    return {"error": f"Unknown tool: {tool_name}"}
 ```
 
 **`agent_turn`**: Executes a single conversation turn with nested subtask execution:
-1. `await call_llm_with_tools(...)` - Call LLM with user message
-2. If tools requested: `await execute_tool(...)` for each tool (which then calls the actual tool task)
-3. `await call_llm_with_tools(...)` again with tool results to generate final response
+1. `await ctx.step(call_llm_with_tools, ...)` - Call LLM with user message
+2. If tools requested: `await ctx.step(execute_tool, ...)` for each tool (which then steps the actual tool task)
+3. `await ctx.step(call_llm_with_tools, ...)` again with tool results to generate final response
 
-This demonstrates **nested subtask calling**: `agent_turn` → `execute_tool` → `get_order_status` (3 levels deep!).
+This demonstrates **nested subtask stepping**: `agent_turn` → `execute_tool` → `get_order_status` (3 levels deep!).
 
 **`multi_turn_conversation`**: Orchestrates multiple conversation turns:
 ```python
 for user_message in messages:
     # SUBTASK CALL: Process each message through agent_turn
-    turn_result = await agent_turn(user_message, conversation_history)
+    turn_result = await ctx.step(agent_turn, user_message, conversation_history)
     conversation_history = turn_result["conversation_history"]
 ```
-This demonstrates **calling subtasks in a loop** to maintain conversation state.
+This demonstrates **stepping subtasks in a loop** to maintain conversation state.
 
 ## Adding New Tools
 
@@ -310,7 +311,7 @@ To add a new tool capability:
 1. **Define the tool function**:
 ```python
 @app.task
-def new_tool(param: str) -> dict:
+def new_tool(ctx: TaskContext, param: str) -> dict:
     """Tool: Description of what this tool does."""
     # Implementation
     return {"result": "data"}
diff --git a/openai-agent/main.py b/openai-agent/main.py
index bfad56f..5419f74 100644
--- a/openai-agent/main.py
+++ b/openai-agent/main.py
@@ -18,7 +18,7 @@
 import os
 from datetime import datetime
 
-from render_sdk import Retry, Workflows
+from render_sdk import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -70,7 +70,7 @@ def create_openai_client() -> "AsyncOpenAI":
 
 
 @app.task
-def get_order_status(order_id: str) -> dict:
+def get_order_status(ctx: TaskContext, order_id: str) -> dict:
     """
     Tool: Look up order status.
 
@@ -109,7 +109,7 @@ def get_order_status(order_id: str) -> dict:
 
 
 @app.task
-def process_refund(order_id: str, reason: str) -> dict:
+def process_refund(ctx: TaskContext, order_id: str, reason: str) -> dict:
     """
     Tool: Process a refund for an order.
 
@@ -142,7 +142,7 @@ def process_refund(order_id: str, reason: str) -> dict:
 
 
 @app.task
-def search_knowledge_base(query: str) -> dict:
+def search_knowledge_base(ctx: TaskContext, query: str) -> dict:
     """
     Tool: Search the knowledge base for information.
 
@@ -194,7 +194,7 @@ def search_knowledge_base(query: str) -> dict:
 
 @app.task
 async def call_llm_with_tools(
-    messages: list[dict], tools: list[dict], model: str = "gpt-4"
+    ctx: TaskContext, messages: list[dict], tools: list[dict], model: str = "gpt-4"
 ) -> dict:
     """
     Call OpenAI with function/tool definitions.
@@ -248,11 +248,12 @@ async def call_llm_with_tools(
 
 
 @app.task
-async def execute_tool(tool_name: str, arguments: dict) -> dict:
+async def execute_tool(ctx: TaskContext, tool_name: str, arguments: dict) -> dict:
     """
     Execute a tool function by name.
 
-    This demonstrates dynamic subtask execution based on agent decisions.
+    This demonstrates dynamic subtask execution via ctx.step, based on
+    agent decisions.
 
     Args:
         tool_name: Name of the tool to execute
@@ -264,30 +265,24 @@ async def execute_tool(tool_name: str, arguments: dict) -> dict:
     logger.info(f"[AGENT] Executing tool: {tool_name}")
     logger.info(f"[AGENT] Arguments: {arguments}")
 
-    # Map tool names to task functions
-    tool_map = {
-        "get_order_status": get_order_status,
-        "process_refund": process_refund,
-        "search_knowledge_base": search_knowledge_base,
-    }
+    # Tool names the agent is allowed to call
+    known_tools = {"get_order_status", "process_refund", "search_knowledge_base"}
 
-    if tool_name not in tool_map:
+    if tool_name not in known_tools:
         logger.error(f"[AGENT] Unknown tool: {tool_name}")
         return {"error": f"Unknown tool: {tool_name}"}
 
-    # Execute the appropriate tool as a subtask
-    tool_function = tool_map[tool_name]
-
     try:
-        # Different tools have different signatures
+        # SUBTASK CALL: ctx.step runs the tool task on its own compute.
+        # Each tool takes different arguments.
         if tool_name == "get_order_status":
-            result = await tool_function(arguments.get("order_id"))
+            result = await ctx.step(get_order_status, arguments.get("order_id"))
         elif tool_name == "process_refund":
-            result = await tool_function(
-                arguments.get("order_id"), arguments.get("reason")
+            result = await ctx.step(
+                process_refund, arguments.get("order_id"), arguments.get("reason")
             )
         elif tool_name == "search_knowledge_base":
-            result = await tool_function(arguments.get("query"))
+            result = await ctx.step(search_knowledge_base, arguments.get("query"))
         else:
             result = {"error": "Tool not implemented"}
 
@@ -301,7 +296,7 @@ async def execute_tool(tool_name: str, arguments: dict) -> dict:
 
 @app.task
 async def agent_turn(
-    user_message: str, conversation_history: list[dict] = None
+    ctx: TaskContext, user_message: str, conversation_history: list[dict] = None
 ) -> dict:
     """
     Execute a single agent turn with tool calling capability.
@@ -411,7 +406,7 @@ async def agent_turn(
     )
 
     # Call LLM
-    llm_response = await call_llm_with_tools(messages, tools)
+    llm_response = await ctx.step(call_llm_with_tools, messages, tools)
 
     # If no tool calls, return the response
     if not llm_response.get("tool_calls"):
@@ -431,7 +426,8 @@ async def agent_turn(
     tool_results = []
 
     for tool_call in llm_response["tool_calls"]:
-        result = await execute_tool(
+        result = await ctx.step(
+            execute_tool,
             tool_call["function"]["name"],
             json.loads(tool_call["function"]["arguments"]),
         )
@@ -453,7 +449,7 @@ async def agent_turn(
         *tool_messages,
     ]
 
-    final_response = await call_llm_with_tools(final_messages, tools)
+    final_response = await ctx.step(call_llm_with_tools, final_messages, tools)
 
     logger.info("[AGENT TURN] Agent turn complete")
 
@@ -469,7 +465,7 @@ async def agent_turn(
 
 
 @app.task
-async def multi_turn_conversation(*messages: str) -> dict:
+async def multi_turn_conversation(ctx: TaskContext, *messages: str) -> dict:
     """
     Run a multi-turn conversation with the agent.
 
@@ -496,7 +492,7 @@ async def multi_turn_conversation(*messages: str) -> dict:
     for i, user_message in enumerate(messages_list, 1):
         logger.info(f"[CONVERSATION] Turn {i}/{len(messages_list)}")
 
-        turn_result = await agent_turn(user_message, conversation_history)
+        turn_result = await ctx.step(agent_turn, user_message, conversation_history)
 
         responses.append(
             {