Add activity_from_node sample demonstrating run_in_workflow

Shows how to call Temporal activities directly from a LangGraph node
using the run_in_workflow=True metadata option.

Author: mfateev

langgraph_samples/README.md

@@ -54,6 +54,7 @@ Each directory contains a complete example with its own README for detailed inst
 | Sample | Description |
 |--------|-------------|
 | [hello_world](./hello_world/) | Simple starter example demonstrating basic plugin setup and graph registration |
+| [activity_from_node](./activity_from_node/) | Calling Temporal activities from a graph node using run_in_workflow |
 | [react_agent](./react_agent/) | ReAct agent pattern with tool calling and multi-step reasoning |
 | [approval_workflow](./approval_workflow/) | Human-in-the-loop with interrupt/resume for approval workflows |
 | [supervisor](./supervisor/) | Multi-agent supervisor pattern coordinating specialized agents |

langgraph_samples/activity_from_node/README.md

@@ -0,0 +1,51 @@
+# Activity from Node
+
+Demonstrates calling Temporal activities directly from a LangGraph node using the `run_in_workflow` feature.
+
+## What This Sample Demonstrates
+
+- **run_in_workflow nodes**: Using `temporal_node_metadata(run_in_workflow=True)` to run a node in the workflow context
+- **Activity orchestration**: Calling multiple Temporal activities from within a graph node
+- **Mixed execution modes**: Combining run_in_workflow nodes with regular activity nodes
+- **Sandbox enforcement**: Node code is sandboxed to ensure deterministic execution
+
+## How It Works
+
+1. **Orchestrator Node**: Runs directly in the workflow (not as an activity) with `run_in_workflow=True`
+   - Calls `validate_data` activity to validate input
+   - Calls `enrich_data` activity to enrich valid data
+   - Implements orchestration logic (conditional activity calls)
+
+2. **Finalize Node**: Runs as a regular Temporal activity (default behavior)
+   - Processes the enriched data
+
+3. **Activities**: Standard Temporal activities called from the orchestrator
+   - `validate_data`: Validates input data
+   - `enrich_data`: Enriches data with additional information
+
+## When to Use run_in_workflow
+
+Use `run_in_workflow=True` when your node needs to:
+- Call Temporal activities, child workflows, or other Temporal operations
+- Use workflow features like timers, signals, or queries
+- Implement complex orchestration logic with multiple activity calls
+
+**Important**: Code in run_in_workflow nodes is sandboxed to ensure deterministic execution. Non-deterministic operations (like `random.randint()`) will be blocked.
+
+## Running the Example
+
+First, start the worker:
+```bash
+uv run langgraph_samples/activity_from_node/run_worker.py
+```
+
+Then, in a separate terminal, run the workflow:
+```bash
+uv run langgraph_samples/activity_from_node/run_workflow.py
+```
+
+## Expected Output
+
+```
+Result: {'data': 'Hello from LangGraph', 'validated': True, 'enriched_data': 'Hello from LangGraph [enriched at activity]', 'final_result': 'Processed: Hello from LangGraph [enriched at activity]'}
+```

langgraph_samples/activity_from_node/__init__.py

@@ -0,0 +1 @@
+"""Activity from Node sample."""

langgraph_samples/activity_from_node/activities.py

@@ -0,0 +1,42 @@
+"""Activity from Node - Activity Definitions.
+
+Activities that are called from a run_in_workflow node.
+"""
+
+from temporalio import activity
+
+
+@activity.defn
+async def validate_data(data: str) -> bool:
+    """Validate the input data.
+
+    In a real application, this could:
+    - Check data format and schema
+    - Verify required fields
+    - Call external validation services
+    """
+    activity.logger.info(f"Validating data: {data}")
+
+    # Simple validation - check if data is non-empty
+    is_valid = bool(data and data.strip())
+
+    activity.logger.info(f"Validation result: {is_valid}")
+    return is_valid
+
+
+@activity.defn
+async def enrich_data(data: str) -> str:
+    """Enrich the input data with additional information.
+
+    In a real application, this could:
+    - Call external APIs for data enrichment
+    - Lookup data from databases
+    - Apply transformations
+    """
+    activity.logger.info(f"Enriching data: {data}")
+
+    # Simple enrichment - add metadata
+    enriched = f"{data} [enriched at activity]"
+
+    activity.logger.info(f"Enriched data: {enriched}")
+    return enriched

langgraph_samples/activity_from_node/graph.py

@@ -0,0 +1,108 @@
+"""Activity from Node - Graph Definition.
+
+This module defines a graph where a node runs in the workflow context
+and calls Temporal activities directly.
+"""
+
+from datetime import timedelta
+from typing import Any
+
+from langgraph.graph import END, START, StateGraph
+from typing_extensions import TypedDict
+
+
+# =============================================================================
+# State Definition
+# =============================================================================
+
+
+class ProcessingState(TypedDict, total=False):
+    """State for the processing graph."""
+
+    data: str
+    validated: bool
+    enriched_data: str
+    final_result: str
+
+
+# =============================================================================
+# Node Functions
+# =============================================================================
+
+
+async def orchestrator_node(state: ProcessingState) -> ProcessingState:
+    """Node that orchestrates multiple activity calls from the workflow.
+
+    This node runs directly in the workflow (run_in_workflow=True) so it can:
+    - Call multiple Temporal activities
+    - Use workflow features like timers, signals, queries
+    - Implement complex orchestration logic
+
+    The node is sandboxed, ensuring deterministic code.
+    """
+    from temporalio import workflow
+
+    data = state.get("data", "")
+
+    # Call validation activity
+    is_valid = await workflow.execute_activity(
+        "validate_data",
+        data,
+        start_to_close_timeout=timedelta(seconds=30),
+    )
+
+    if not is_valid:
+        return {"validated": False, "final_result": "Validation failed"}
+
+    # Call enrichment activity
+    enriched = await workflow.execute_activity(
+        "enrich_data",
+        data,
+        start_to_close_timeout=timedelta(seconds=30),
+    )
+
+    return {"validated": True, "enriched_data": enriched}
+
+
+def finalize_node(state: ProcessingState) -> ProcessingState:
+    """Final processing node - runs as a regular activity.
+
+    This demonstrates mixing run_in_workflow nodes with regular activity nodes.
+    """
+    if not state.get("validated"):
+        return state
+
+    enriched = state.get("enriched_data", "")
+    return {"final_result": f"Processed: {enriched}"}
+
+
+# =============================================================================
+# Graph Builder
+# =============================================================================
+
+
+def build_activity_from_node_graph() -> Any:
+    """Build a graph with a node that calls activities from the workflow.
+
+    The orchestrator node uses run_in_workflow=True to execute directly
+    in the workflow context, allowing it to call Temporal activities.
+    """
+    from temporalio.contrib.langgraph import temporal_node_metadata
+
+    graph = StateGraph(ProcessingState)
+
+    # Orchestrator runs in workflow to call activities
+    graph.add_node(
+        "orchestrator",
+        orchestrator_node,
+        metadata=temporal_node_metadata(run_in_workflow=True),
+    )
+
+    # Finalize runs as a regular activity
+    graph.add_node("finalize", finalize_node)
+
+    graph.add_edge(START, "orchestrator")
+    graph.add_edge("orchestrator", "finalize")
+    graph.add_edge("finalize", END)
+
+    return graph.compile()

langgraph_samples/activity_from_node/run_worker.py

@@ -0,0 +1,43 @@
+"""Worker for the Activity from Node example.
+
+Starts a Temporal worker that can execute ActivityFromNodeWorkflow.
+"""
+
+import asyncio
+
+from temporalio.client import Client
+from temporalio.contrib.langgraph import LangGraphPlugin
+from temporalio.envconfig import ClientConfig
+from temporalio.worker import Worker
+
+from langgraph_samples.activity_from_node.activities import enrich_data, validate_data
+from langgraph_samples.activity_from_node.graph import build_activity_from_node_graph
+from langgraph_samples.activity_from_node.workflow import ActivityFromNodeWorkflow
+
+
+async def main() -> None:
+    # Create the plugin with our graph registered
+    plugin = LangGraphPlugin(
+        graphs={"activity_from_node_graph": build_activity_from_node_graph},
+    )
+
+    # Connect to Temporal with the plugin
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config, plugins=[plugin])
+
+    # Create and run the worker
+    # Note: We register our custom activities alongside LangGraph's auto-registered ones
+    worker = Worker(
+        client,
+        task_queue="langgraph-activity-from-node",
+        workflows=[ActivityFromNodeWorkflow],
+        activities=[validate_data, enrich_data],
+    )
+
+    print("Worker started. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_samples/activity_from_node/run_workflow.py

@@ -0,0 +1,33 @@
+"""Run the Activity from Node workflow.
+
+Starts a workflow execution and waits for the result.
+"""
+
+import asyncio
+import uuid
+
+from temporalio.client import Client
+from temporalio.envconfig import ClientConfig
+
+from langgraph_samples.activity_from_node.workflow import ActivityFromNodeWorkflow
+
+
+async def main() -> None:
+    # Connect to Temporal
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    client = await Client.connect(**config)
+
+    # Run the workflow
+    result = await client.execute_workflow(
+        ActivityFromNodeWorkflow.run,
+        "Hello from LangGraph",
+        id=f"activity-from-node-{uuid.uuid4()}",
+        task_queue="langgraph-activity-from-node",
+    )
+
+    print(f"Result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

langgraph_samples/activity_from_node/workflow.py

@@ -0,0 +1,39 @@
+"""Activity from Node - Workflow Definition.
+
+Demonstrates calling Temporal activities from a LangGraph node.
+
+Note: This module only contains the workflow definition. The graph and
+activities are defined separately and imported only by the worker.
+"""
+
+from typing import Any
+
+from temporalio import workflow
+from temporalio.contrib.langgraph import compile
+
+
+@workflow.defn
+class ActivityFromNodeWorkflow:
+    """Workflow that runs a graph with activity-calling nodes.
+
+    This demonstrates:
+    - Using run_in_workflow=True for nodes that need workflow context
+    - Calling Temporal activities from within graph nodes
+    - Mixing run_in_workflow nodes with regular activity nodes
+    """
+
+    @workflow.run
+    async def run(self, data: str) -> dict[str, Any]:
+        """Run the processing graph.
+
+        Args:
+            data: The input data to process.
+
+        Returns:
+            The final state with processing results.
+        """
+        app = compile("activity_from_node_graph")
+
+        result = await app.ainvoke({"data": data})
+
+        return result