chore: use run_context instead of session_state in workflow code

examples/agent-os/workflow/customer-research-workflow-parallel.mdx

@@ -318,11 +318,12 @@ def set_session_state_step(
 
 
 async def customer_profile_research_step(
-    step_input: StepInput, session_state: dict
+    step_input: StepInput, run_context: RunContext
 ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]:
     """
     Conduct customer profile research with session state tracking
     """
+    session_state = run_context.session_state
     customer_query = step_input.input
     previous_content = step_input.previous_step_content
 
@@ -388,11 +389,12 @@ async def customer_profile_research_step(
 
 
 async def customer_biz_goals_research_step(
-    step_input: StepInput, session_state: dict
+    step_input: StepInput, run_context: RunContext
 ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]:
     """
     Conduct business goals research with session state tracking
     """
+    session_state = run_context.session_state
     customer_query = step_input.input
 
     # Update session state
@@ -457,11 +459,12 @@ async def customer_biz_goals_research_step(
 
 
 async def web_intelligence_research_step(
-    step_input: StepInput, session_state: dict
+    step_input: StepInput, run_context: RunContext
 ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]:
     """
     Conduct web intelligence research with session state tracking
     """
+    session_state = run_context.session_state
     customer_query = step_input.input
 
     # Update session state
@@ -526,11 +529,12 @@ async def web_intelligence_research_step(
 
 
 async def customer_report_consolidation_step(
-    step_input: StepInput, session_state: dict
+    step_input: StepInput, run_context: RunContext
 ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]:
     """
     Consolidate all research findings into comprehensive customer report
     """
+    session_state = run_context.session_state
     customer_query = step_input.input
 
     # Gather all research findings from session state
@@ -626,11 +630,12 @@ async def customer_report_consolidation_step(
 
 
 async def task_recommender_step(
-    step_input: StepInput, session_state: dict
+    step_input: StepInput, run_context: RunContext
 ) -> AsyncIterator[Union[WorkflowRunOutputEvent, StepOutput]]:
     """
     Generate actionable task recommendations based on consolidated research
     """
+    session_state = run_context.session_state
     customer_query = step_input.input
 
     research_data = session_state["customer_research"]

examples/workflows/advanced-concepts/session-state/state-in-condition.mdx

@@ -14,6 +14,7 @@ Demonstrates using workflow session state in a `Condition` evaluator and executo
 
 from agno.agent import Agent
 from agno.models.openai import OpenAIChat
+from agno.run import RunContext
 from agno.workflow.condition import Condition
 from agno.workflow.step import Step, StepInput, StepOutput
 from agno.workflow.workflow import Workflow
@@ -22,22 +23,22 @@ from agno.workflow.workflow import Workflow
 # ---------------------------------------------------------------------------
 # Define Session-State Functions
 # ---------------------------------------------------------------------------
-def check_user_has_context(step_input: StepInput, session_state: dict) -> bool:
+def check_user_has_context(step_input: StepInput, run_context: RunContext) -> bool:
     print("\n=== Evaluating Condition ===")
-    print(f"User ID: {session_state.get('current_user_id')}")
-    print(f"Session ID: {session_state.get('current_session_id')}")
-    print(f"Has been greeted: {session_state.get('has_been_greeted', False)}")
+    print(f"User ID: {run_context.session_state.get('current_user_id')}")
+    print(f"Session ID: {run_context.session_state.get('current_session_id')}")
+    print(f"Has been greeted: {run_context.session_state.get('has_been_greeted', False)}")
 
-    return session_state.get("has_been_greeted", False)
+    return run_context.session_state.get("has_been_greeted", False)
 
 
-def mark_user_as_greeted(step_input: StepInput, session_state: dict) -> StepOutput:
+def mark_user_as_greeted(step_input: StepInput, run_context: RunContext) -> StepOutput:
     print("\n=== Marking User as Greeted ===")
-    session_state["has_been_greeted"] = True
-    session_state["greeting_count"] = session_state.get("greeting_count", 0) + 1
+    run_context.session_state["has_been_greeted"] = True
+    run_context.session_state["greeting_count"] = run_context.session_state.get("greeting_count", 0) + 1
 
     return StepOutput(
-        content=f"User has been greeted. Total greetings: {session_state['greeting_count']}"
+        content=f"User has been greeted. Total greetings: {run_context.session_state['greeting_count']}"
     )
 
 
@@ -67,10 +68,10 @@ workflow = Workflow(
         Condition(
             name="Check If New User",
             description="Check if this is a new user who needs greeting",
-            evaluator=lambda step_input, session_state: (
+            evaluator=lambda step_input, run_context: (
                 not check_user_has_context(
                     step_input,
-                    session_state,
+                    run_context,
                 )
             ),
             steps=[

examples/workflows/advanced-concepts/session-state/state-in-function.mdx

@@ -18,6 +18,7 @@ from agno.agent import Agent
 from agno.db.in_memory import InMemoryDb
 from agno.db.sqlite import SqliteDb
 from agno.models.openai import OpenAIChat
+from agno.run import RunContext
 from agno.run.workflow import WorkflowRunOutputEvent
 from agno.team import Team
 from agno.tools.hackernews import HackerNewsTools
@@ -77,8 +78,9 @@ research_team = Team(
 # ---------------------------------------------------------------------------
 def custom_content_planning_function(
     step_input: StepInput,
-    session_state: dict,
+    run_context: RunContext,
 ) -> StepOutput:
+    session_state = run_context.session_state
     message = step_input.input
     previous_step_content = step_input.previous_step_content
 
@@ -150,7 +152,8 @@ def custom_content_planning_function(
         )
 
 
-def content_summary_function(step_input: StepInput, session_state: dict) -> StepOutput:
+def content_summary_function(step_input: StepInput, run_context: RunContext) -> StepOutput:
+    session_state = run_context.session_state
     if "content_plans" not in session_state or not session_state["content_plans"]:
         return StepOutput(
             content="No content plans found in session state.", success=False
@@ -184,8 +187,9 @@ def content_summary_function(step_input: StepInput, session_state: dict) -> Step
 
 def custom_content_planning_function_stream(
     step_input: StepInput,
-    session_state: dict,
+    run_context: RunContext,
 ) -> Iterator[Union[WorkflowRunOutputEvent, StepOutput]]:
+    session_state = run_context.session_state
     message = step_input.input
     previous_step_content = step_input.previous_step_content
 
@@ -267,8 +271,9 @@ def custom_content_planning_function_stream(
 
 def content_summary_function_stream(
     step_input: StepInput,
-    session_state: dict,
+    run_context: RunContext,
 ) -> Iterator[StepOutput]:
+    session_state = run_context.session_state
     plans = session_state["content_plans"]
 
     summary = f"""

examples/workflows/advanced-concepts/session-state/state-in-router.mdx

@@ -27,18 +27,18 @@ from agno.workflow.workflow import Workflow
 # ---------------------------------------------------------------------------
 # Define Router Functions (Preference-Based Routing)
 # ---------------------------------------------------------------------------
-def route_based_on_user_preference(step_input: StepInput, session_state: dict) -> Step:
+def route_based_on_user_preference(step_input: StepInput, run_context: RunContext) -> Step:
     print("\n=== Routing Decision ===")
-    print(f"User ID: {session_state.get('current_user_id')}")
-    print(f"Session ID: {session_state.get('current_session_id')}")
+    print(f"User ID: {run_context.session_state.get('current_user_id')}")
+    print(f"Session ID: {run_context.session_state.get('current_session_id')}")
 
-    user_preference = session_state.get("agent_preference", "general")
-    interaction_count = session_state.get("interaction_count", 0)
+    user_preference = run_context.session_state.get("agent_preference", "general")
+    interaction_count = run_context.session_state.get("interaction_count", 0)
 
     print(f"User Preference: {user_preference}")
     print(f"Interaction Count: {interaction_count}")
 
-    session_state["interaction_count"] = interaction_count + 1
+    run_context.session_state["interaction_count"] = interaction_count + 1
 
     if user_preference == "technical":
         print("Routing to Technical Expert")
@@ -55,18 +55,18 @@ def route_based_on_user_preference(step_input: StepInput, session_state: dict) -
     return general_step
 
 
-def set_user_preference(step_input: StepInput, session_state: dict) -> StepOutput:
+def set_user_preference(step_input: StepInput, run_context: RunContext) -> StepOutput:
     print("\n=== Setting User Preference ===")
-    interaction_count = session_state.get("interaction_count", 0)
+    interaction_count = run_context.session_state.get("interaction_count", 0)
 
     if interaction_count % 3 == 1:
-        session_state["agent_preference"] = "technical"
+        run_context.session_state["agent_preference"] = "technical"
         preference = "technical"
     elif interaction_count % 3 == 2:
-        session_state["agent_preference"] = "friendly"
+        run_context.session_state["agent_preference"] = "friendly"
         preference = "friendly"
     else:
-        session_state["agent_preference"] = "general"
+        run_context.session_state["agent_preference"] = "general"
         preference = "general"
 
     print(f"Set preference to: {preference}")

examples/workflows/cel-expressions/condition/cel-session-state.mdx

@@ -20,6 +20,7 @@ Requirements:
 
 from agno.agent import Agent
 from agno.models.openai import OpenAIChat
+from agno.run import RunContext
 from agno.workflow import (
     CEL_AVAILABLE,
     Condition,
@@ -40,19 +41,19 @@ if not CEL_AVAILABLE:
 # ---------------------------------------------------------------------------
 # Define Helpers
 # ---------------------------------------------------------------------------
-def increment_retry_count(step_input: StepInput, session_state: dict) -> StepOutput:
+def increment_retry_count(step_input: StepInput, run_context: RunContext) -> StepOutput:
     """Increment retry count in session state."""
-    current_count = session_state.get("retry_count", 0)
-    session_state["retry_count"] = current_count + 1
+    current_count = run_context.session_state.get("retry_count", 0)
+    run_context.session_state["retry_count"] = current_count + 1
     return StepOutput(
-        content=f"Retry count incremented to {session_state['retry_count']}",
+        content=f"Retry count incremented to {run_context.session_state['retry_count']}",
         success=True,
     )
 
 
-def reset_retry_count(step_input: StepInput, session_state: dict) -> StepOutput:
+def reset_retry_count(step_input: StepInput, run_context: RunContext) -> StepOutput:
     """Reset retry count in session state."""
-    session_state["retry_count"] = 0
+    run_context.session_state["retry_count"] = 0
     return StepOutput(content="Retry count reset to 0", success=True)
 
 

state/workflows/access-session-state-in-condition-evaluator-function.mdx

@@ -6,10 +6,14 @@ mode: wide
 ---
 
 This example shows:
-1. Using `session_state` in a Condition evaluator function
-2. Reading and modifying `session_state` based on condition logic
-3. Accessing `user_id` and `session_id` from `session_state`
-4. Making conditional decisions based on `session_state`
+1. Using `run_context.session_state` in a Condition evaluator function
+2. Reading and modifying session state based on condition logic
+3. Accessing `user_id` and `session_id` from session state
+4. Making conditional decisions based on session state
+
+<Note>
+Accepting a `session_state` parameter directly is deprecated. Accept `run_context: RunContext` and use `run_context.session_state` instead.
+</Note>
 
 <Steps>
 <Step title="Create a Python file">
@@ -18,39 +22,40 @@ This example shows:
 from agno.agent import Agent
 from agno.db.in_memory import InMemoryDb
 from agno.models.openai import OpenAIResponses
+from agno.run import RunContext
 from agno.workflow.condition import Condition
 from agno.workflow.step import Step, StepInput, StepOutput
 from agno.workflow.workflow import Workflow
 
 
-def check_user_has_context(step_input: StepInput, session_state: dict) -> bool:
+def check_user_has_context(step_input: StepInput, run_context: RunContext) -> bool:
     """
     Condition evaluator that checks if user has been greeted before.
 
     Args:
         step_input: The input for this step (contains workflow context)
-        session_state: The shared session state
+        run_context: The run context holding the workflow session state
 
     Returns:
         bool: True if user has context, False otherwise
     """
     print("\n=== Evaluating Condition ===")
-    print(f"User ID: {session_state.get('current_user_id')}")
-    print(f"Session ID: {session_state.get('current_session_id')}")
-    print(f"Has been greeted: {session_state.get('has_been_greeted', False)}")
+    print(f"User ID: {run_context.session_state.get('current_user_id')}")
+    print(f"Session ID: {run_context.session_state.get('current_session_id')}")
+    print(f"Has been greeted: {run_context.session_state.get('has_been_greeted', False)}")
 
     # Check if user has been greeted before
-    return session_state.get("has_been_greeted", False)
+    return run_context.session_state.get("has_been_greeted", False)
 
 
-def mark_user_as_greeted(step_input: StepInput, session_state: dict) -> StepOutput:
+def mark_user_as_greeted(step_input: StepInput, run_context: RunContext) -> StepOutput:
     """Custom function that marks user as greeted in session state."""
     print("\n=== Marking User as Greeted ===")
-    session_state["has_been_greeted"] = True
-    session_state["greeting_count"] = session_state.get("greeting_count", 0) + 1
+    run_context.session_state["has_been_greeted"] = True
+    run_context.session_state["greeting_count"] = run_context.session_state.get("greeting_count", 0) + 1
 
     return StepOutput(
-        content=f"User has been greeted. Total greetings: {session_state['greeting_count']}"
+        content=f"User has been greeted. Total greetings: {run_context.session_state['greeting_count']}"
     )
 
 
@@ -79,8 +84,8 @@ workflow = Workflow(
             name="Check If New User",
             description="Check if this is a new user who needs greeting",
             # Condition returns True if user has context, so we negate it
-            evaluator=lambda step_input, session_state: not check_user_has_context(
-                step_input, session_state
+            evaluator=lambda step_input, run_context: not check_user_has_context(
+                step_input, run_context
             ),
             steps=[
                 # Only execute these steps for new users