Dynamic State

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

1uv pip install -U kern-ai openai

1python dynamic_session_state.py