From ec41765879207158b5a602a59096de35530effe1 Mon Sep 17 00:00:00 2001 From: kausmeows Date: Thu, 16 Jul 2026 18:44:05 +0530 Subject: [PATCH] chore: use run_context instead of session_state in workflow code --- .../customer-research-workflow-parallel.mdx | 15 +++++--- .../session-state/state-in-condition.mdx | 23 ++++++------ .../session-state/state-in-function.mdx | 13 +++++-- .../session-state/state-in-router.mdx | 22 +++++------ .../condition/cel-session-state.mdx | 13 ++++--- ...-state-in-condition-evaluator-function.mdx | 37 +++++++++++-------- ...sion-state-in-router-selector-function.mdx | 31 +++++++++------- state/workflows/overview.mdx | 14 ++++--- 8 files changed, 97 insertions(+), 71 deletions(-) diff --git a/examples/agent-os/workflow/customer-research-workflow-parallel.mdx b/examples/agent-os/workflow/customer-research-workflow-parallel.mdx index 04c6b47f8..8968d63c3 100644 --- a/examples/agent-os/workflow/customer-research-workflow-parallel.mdx +++ b/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"] diff --git a/examples/workflows/advanced-concepts/session-state/state-in-condition.mdx b/examples/workflows/advanced-concepts/session-state/state-in-condition.mdx index 895038de3..bb2814990 100644 --- a/examples/workflows/advanced-concepts/session-state/state-in-condition.mdx +++ b/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=[ diff --git a/examples/workflows/advanced-concepts/session-state/state-in-function.mdx b/examples/workflows/advanced-concepts/session-state/state-in-function.mdx index 047b49513..9bff70f65 100644 --- a/examples/workflows/advanced-concepts/session-state/state-in-function.mdx +++ b/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""" diff --git a/examples/workflows/advanced-concepts/session-state/state-in-router.mdx b/examples/workflows/advanced-concepts/session-state/state-in-router.mdx index fdec89b91..4e6be68d7 100644 --- a/examples/workflows/advanced-concepts/session-state/state-in-router.mdx +++ b/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}") diff --git a/examples/workflows/cel-expressions/condition/cel-session-state.mdx b/examples/workflows/cel-expressions/condition/cel-session-state.mdx index 7179f8f4f..21808c2a6 100644 --- a/examples/workflows/cel-expressions/condition/cel-session-state.mdx +++ b/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) diff --git a/state/workflows/access-session-state-in-condition-evaluator-function.mdx b/state/workflows/access-session-state-in-condition-evaluator-function.mdx index 107fa2477..fbbf0c0a4 100644 --- a/state/workflows/access-session-state-in-condition-evaluator-function.mdx +++ b/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 + + +Accepting a `session_state` parameter directly is deprecated. Accept `run_context: RunContext` and use `run_context.session_state` instead. + @@ -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 diff --git a/state/workflows/access-session-state-in-router-selector-function.mdx b/state/workflows/access-session-state-in-router-selector-function.mdx index 72a5c5044..35fdbbef8 100644 --- a/state/workflows/access-session-state-in-router-selector-function.mdx +++ b/state/workflows/access-session-state-in-router-selector-function.mdx @@ -5,7 +5,11 @@ description: Access session state in the selector function of a router step. mode: wide --- -The `Router` selector reads and updates `session_state` to choose a step from user preferences and prior interactions. +The `Router` selector reads and updates `run_context.session_state` to choose a step from user preferences and prior interactions. + + +Accepting a `session_state` parameter directly is deprecated. Accept `run_context: RunContext` and use `run_context.session_state` instead. + @@ -14,35 +18,36 @@ The `Router` selector reads and updates `session_state` to choose a step from us 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.router import Router from agno.workflow.step import Step, StepInput, StepOutput from agno.workflow.workflow import Workflow -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: """ Router selector that chooses an agent based on user preferences in session_state. Args: step_input: The input for this step (contains user query) - session_state: The workflow session state dict + run_context: The run context holding the workflow session state Returns: Step: The step to execute based on user preference """ 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')}") # Get user preference from session state - 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}") # Update interaction count - session_state["interaction_count"] = interaction_count + 1 + run_context.session_state["interaction_count"] = interaction_count + 1 # Route based on preference if user_preference == "technical": @@ -61,22 +66,22 @@ 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: """Custom function that sets user preference based on onboarding.""" print("\n=== Setting User Preference ===") # In a real scenario, this would analyze the user's response # For demo purposes, we'll set it based on interaction count - 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}") diff --git a/state/workflows/overview.mdx b/state/workflows/overview.mdx index 527839fb9..c9cf83c91 100644 --- a/state/workflows/overview.mdx +++ b/state/workflows/overview.mdx @@ -217,11 +217,11 @@ def custom_function_step(step_input: StepInput, run_context: RunContext): See [examples](/state/workflows/access-session-state-in-custom-python-function-step) for more details. -The session state is also available as a `session_state` dict parameter in the evaluator and selector functions of the `Condition` and `Router` steps: +The `run_context` parameter is also injected into the evaluator and selector functions of the `Condition` and `Router` steps: ```python -def evaluator_function(step_input: StepInput, session_state: dict) -> bool: - return session_state["test"] == "test_1" +def evaluator_function(step_input: StepInput, run_context: RunContext) -> bool: + return run_context.session_state["test"] == "test_1" condition_step = Condition( name="condition_step", @@ -231,8 +231,8 @@ condition_step = Condition( ``` ```python -def selector_function(step_input: StepInput, session_state: dict) -> Step: - if session_state["test"] == "test_1": +def selector_function(step_input: StepInput, run_context: RunContext) -> Step: + if run_context.session_state["test"] == "test_1": return step_1 return step_2 @@ -243,6 +243,10 @@ router_step = Router( ) ``` + +Accepting a `session_state` dict parameter directly in custom step functions, evaluators, and selectors is deprecated. Accept `run_context: RunContext` and use `run_context.session_state` instead. + + See example of [Session State in Condition Evaluator Function](/state/workflows/access-session-state-in-condition-evaluator-function) and [Session State in Router Selector Function](/state/workflows/access-session-state-in-router-selector-function) for more details.