Dynamic Session State

1"""
2Dynamic Session State
3=============================
4
5Dynamic Session State.
6"""
7
8import json
9from typing import Any, Dict
10
11from kern.agent import Agent
12from kern.db.in_memory import InMemoryDb
13from kern.models.openai import OpenAIResponses
14from kern.run import RunContext
15from kern.tools.toolkit import Toolkit
16from kern.utils.log import log_info, log_warning
17
18
19class CustomerDBTools(Toolkit):
20    def __init__(self, *args, **kwargs):
21        super().__init__(*args, **kwargs)
22        self.register(self.process_customer_request)
23
24    def process_customer_request(
25        self,
26        agent: Agent,
27        customer_id: str,
28        action: str = "retrieve",
29        name: str = "John Doe",
30    ):
31        log_warning("Tool called, this shouldn't happen.")
32        return "This should not be seen."
33
34
35def customer_management_hook(run_context: RunContext, arguments: Dict[str, Any]):
36    if run_context.session_state is None:
37        run_context.session_state = {}
38
39    action = arguments.get("action", "retrieve")
40    cust_id = arguments.get("customer_id")
41    name = arguments.get("name", None)
42
43    if not cust_id:
44        raise ValueError("customer_id is required.")
45
46    if action == "create":
47        run_context.session_state["customer_profiles"][cust_id] = {"name": name}
48        log_info(f"Hook: UPDATED session_state for customer '{cust_id}'.")
49        return f"Success! Customer {cust_id} has been created."
50
51    if action == "retrieve":
52        profile = run_context.session_state.get("customer_profiles", {}).get(cust_id)
53        if profile:
54            log_info(f"Hook: FOUND customer '{cust_id}' in session_state.")
55            return f"Profile for {cust_id}: {json.dumps(profile)}"
56        else:
57            raise ValueError(f"Customer '{cust_id}' not found.")
58
59    log_info(f"Session state: {run_context.session_state}")
60
61
62# ---------------------------------------------------------------------------
63# Create Agent
64# ---------------------------------------------------------------------------
65def run_test():
66    # ---------------------------------------------------------------------------
67    # Create Agent
68    # ---------------------------------------------------------------------------
69
70    agent = Agent(
71        model=OpenAIResponses(id="gpt-5.2"),
72        tools=[CustomerDBTools()],
73        tool_hooks=[customer_management_hook],
74        session_state={"customer_profiles": {"123": {"name": "Jane Doe"}}},
75        instructions="Your profiles: {customer_profiles}. Use `process_customer_request`. Use either create or retrieve as action for the tool.",
76        resolve_in_context=True,
77        db=InMemoryDb(),
78    )
79
80    prompt = "First, create customer 789 named 'Tom'. Then, retrieve Tom's profile. Step by step."
81    log_info(f" Prompting: '{prompt}'")
82    agent.print_response(prompt, stream=False)
83
84    log_info("\n--- TEST ANALYSIS ---")
85    log_info(
86        "Check logs for the second tool call. The system prompt will NOT contain customer '789'."
87    )
88
89
90# ---------------------------------------------------------------------------
91# Run Agent
92# ---------------------------------------------------------------------------
93if __name__ == "__main__":
94    run_test()

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/05_state_and_session
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python dynamic_session_state.py