update to new sdk

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

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

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")
 

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

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