# Session State Hooks (/examples/agents/hooks/session-state-hooks) Example demonstrating how to use a pre\_hook to update the session\_state. ```python title="session_state_hooks.py" """ Session State Hooks ============================= Example demonstrating how to use a pre_hook to update the session_state. """ from typing import List from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.run import RunContext from agno.run.agent import RunInput from pydantic import BaseModel, Field class ConversationTopics(BaseModel): topics: List[str] = Field(description="Topics present in the user messages") # This will be our pre-hook function def track_conversation_topics(run_context: RunContext, run_input: RunInput) -> None: """Simple pre-hook function to track conversation topics in the session state""" # Initialize the session state if it doesn't exist yet if run_context.session_state is None: run_context.session_state = {"topics": []} elif run_context.session_state.get("topics") is None: run_context.session_state["topics"] = [] # Setup an Agent to get the topics discussed in the conversation # --------------------------------------------------------------------------- # Create Agent # --------------------------------------------------------------------------- topics_analyzer_agent = Agent( name="Topics Analyzer", model=OpenAIResponses(id="gpt-5-mini"), instructions=[ "Your task is to analyze a user query and extract the topics." "You will be presented with a user message sent to an agent." "You need to extract the topics present in the user message." "Be concise and brief. Topics should be one or two words, and only want the one or two main topics." "Respond just with the list of topics, no other text or explanation." ], output_schema=ConversationTopics, ) # Run the Agent to get the topics discussed in the conversation response = topics_analyzer_agent.run( input=f"Extract the topics present in the following user message: {run_input.input_content}" ) # Update the session state to track the topics discussed in the conversation run_context.session_state["topics"].extend(response.content.topics) # type: ignore # --------------------------------------------------------------------------- # Create Agent # --------------------------------------------------------------------------- # Create a simple agent and equip it with our pre-hook agent = Agent( name="Simple Agent", model=OpenAIResponses(id="gpt-5-mini"), pre_hooks=[track_conversation_topics], db=SqliteDb(db_file="test.db"), ) # --------------------------------------------------------------------------- # Run Agent # --------------------------------------------------------------------------- if __name__ == "__main__": agent.print_response( input="I want to know more about AI Agents.", session_id="topics_analyzer_session", ) print( f"Current session state, after the first run: {agent.get_session_state(session_id='topics_analyzer_session')}" ) agent.print_response( input="I also want to know more about Agno, the framework to build AI Agents.", session_id="topics_analyzer_session", ) print( f"Current session state, after the second run: {agent.get_session_state(session_id='topics_analyzer_session')}" ) ``` Guardrail `InputCheckError` and `OutputCheckError` become failed run outputs. They are not caught by the source's `try`/`except` around `print_response()` or `aprint_response()`. Use `run()` or `arun()` and check the returned status before displaying content or declaring success. Add this helper after your imports: ```python from agno.run import RunStatus def show_checked_response(response) -> None: if response.status != RunStatus.completed: print(f"Run rejected or failed ({response.status.value}).") return print(response.content) ``` A generic failed status can also indicate a provider error. Nonstream run outputs do not expose a `check_trigger` field. Output rejected by a post-hook can remain in the run record; this helper withholds it from the display. Streaming content may already have been emitted before the post-hook runs. ### Handle a failed topic analysis [#handle-a-failed-topic-analysis] Also import `CheckTrigger` and `InputCheckError` from `agno.exceptions`. Immediately before extending `run_context.session_state["topics"]`, insert this check. This version rejects the outer run if topic analysis fails, so it cannot silently claim to have updated topics. ```python if response.status != RunStatus.completed or not isinstance(response.content, ConversationTopics): raise InputCheckError( "The validator did not return a valid result.", check_trigger=CheckTrigger.INPUT_NOT_ALLOWED, ) ``` After adding the helper and typed topic check, replace the final entrypoint with: ```python if __name__ == "__main__": for prompt in ( "I want to know more about AI Agents.", "I also want to know more about Agno, the framework to build AI Agents.", ): response = agent.run(input=prompt, session_id="topics_analyzer_session") show_checked_response(response) if response.status == RunStatus.completed: print(agent.get_session_state(session_id="topics_analyzer_session")) ``` ## Run the Example [#run-the-example] Mac Windows ```bash uv venv --python 3.12 source .venv/bin/activate ``` ```bash uv venv --python 3.12 .venv\Scripts\activate ``` ```bash uv pip install -U agno openai sqlalchemy ``` Mac/Linux Windows ```bash export OPENAI_API_KEY="your_openai_api_key_here" ``` ```bash $Env:OPENAI_API_KEY="your_openai_api_key_here" ``` Add the status helper and apply the replacements described above. Keep the original imports and agent definitions that the replacement uses. Save the adapted code as `session_state_hooks.py`, then run: ```bash python session_state_hooks.py ``` Full source: [cookbook/02\_agents/09\_hooks/session\_state\_hooks.py](https://github.com/agno-agi/agno/blob/8f36eaf2d18e91afa7b327eec66a3cd3685dcb87/cookbook/02_agents/09_hooks/session_state_hooks.py)