State in Custom Function

1from kern.agent import Agent
2from kern.db.sqlite import SqliteDb
3from kern.models.openai import OpenAIResponses
4from kern.run import RunContext
5from kern.team import Team
6from kern.tools.hackernews import HackerNewsTools
7from kern.tools.yfinance import YFinanceTools
8from kern.workflow.step import Step, StepInput, StepOutput
9from kern.workflow.workflow import Workflow
10
11# Define agents
12hackernews_agent = Agent(
13    name="Hackernews Agent",
14    model=OpenAIResponses(id="gpt-5.2"),
15    tools=[HackerNewsTools()],
16    instructions="Extract key insights and content from Hackernews posts",
17)
18
19finance_agent = Agent(
20    name="Finance Agent",
21    model=OpenAIResponses(id="gpt-5.2"),
22    tools=[YFinanceTools()],
23    instructions="Get financial data and market trends",
24)
25
26# Define research team for complex analysis
27research_team = Team(
28    name="Research Team",
29    model=OpenAIResponses(id="gpt-5.2"),
30    members=[hackernews_agent, finance_agent],
31    instructions="Analyze content and create comprehensive social media strategy",
32)
33
34content_planner = Agent(
35    name="Content Planner",
36    model=OpenAIResponses(id="gpt-5.2"),
37    instructions=[
38        "Plan a content schedule over 4 weeks for the provided topic and research content",
39        "Ensure that I have posts for 3 posts per week",
40    ],
41)
42
43
44def custom_content_planning_function(
45    step_input: StepInput, run_context: RunContext
46) -> StepOutput:
47    """
48    Custom function that does intelligent content planning with context awareness
49    and maintains a content plan history in session_state
50    """
51    message = step_input.input
52    previous_step_content = step_input.previous_step_content
53
54    # Initialize content history if not present
55    if "content_plans" not in run_context.session_state:
56        run_context.session_state["content_plans"] = []
57
58    if "plan_counter" not in run_context.session_state:
59        run_context.session_state["plan_counter"] = 0
60
61    # Increment plan counter
62    run_context.session_state["plan_counter"] += 1
63    current_plan_id = run_context.session_state["plan_counter"]
64
65    # Create intelligent planning prompt
66    planning_prompt = f"""
67        STRATEGIC CONTENT PLANNING REQUEST:
68
69        Core Topic: {message}
70        Plan ID: #{current_plan_id}
71
72        Research Results: {previous_step_content[:500] if previous_step_content else "No research results"}
73
74        Previous Plans Count: {len(run_context.session_state["content_plans"])}
75
76        Planning Requirements:
77        1. Create a comprehensive content strategy based on the research
78        2. Leverage the research findings effectively
79        3. Identify content formats and channels
80        4. Provide timeline and priority recommendations
81        5. Include engagement and distribution strategies
82
83        Please create a detailed, actionable content plan.
84    """
85
86    try:
87        response = content_planner.run(planning_prompt)
88
89        # Store this plan in session state
90        plan_data = {
91            "id": current_plan_id,
92            "topic": message,
93            "content": response.content,
94            "timestamp": f"Plan #{current_plan_id}",
95            "has_research": bool(previous_step_content),
96        }
97        run_context.session_state["content_plans"].append(plan_data)
98
99        enhanced_content = f"""
100            ## Strategic Content Plan #{current_plan_id}
101
102            **Planning Topic:** {message}
103
104            **Research Integration:** {"Research-based" if previous_step_content else "No research foundation"}
105            **Total Plans Created:** {len(run_context.session_state["content_plans"])}
106
107            **Content Strategy:**
108            {response.content}
109
110            **Custom Planning Enhancements:**
111            - Research Integration: {"High" if previous_step_content else "Baseline"}
112            - Strategic Alignment: Optimized for multi-channel distribution
113            - Execution Ready: Detailed action items included
114            - Session History: {len(run_context.session_state["content_plans"])} plans stored
115
116            **Plan ID:** #{current_plan_id}
117        """.strip()
118
119        return StepOutput(content=enhanced_content)
120
121    except Exception as e:
122        return StepOutput(
123            content=f"Custom content planning failed: {str(e)}",
124            success=False,
125        )
126
127
128def content_summary_function(step_input: StepInput, run_context: RunContext) -> StepOutput:
129    """
130    Custom function that summarizes all content plans created in the session
131    """
132    if run_context.session_state is None or run_context.session_state.get("content_plans") is None:
133        return StepOutput(
134            content="No content plans found in session state.", success=False
135        )
136
137    plans = run_context.session_state["content_plans"]
138    summary = f"""
139        ## Content Planning Session Summary
140
141        **Total Plans Created:** {len(plans)}
142        **Session Statistics:**
143        - Plans with research: {len([p for p in plans if p["has_research"]])}
144        - Plans without research: {len([p for p in plans if not p["has_research"]])}
145
146        **Plan Overview:**
147    """
148
149    for plan in plans:
150        summary += f"""
151
152        ### Plan #{plan["id"]} - {plan["topic"]}
153        - Research Available: {"Yes" if plan["has_research"] else "No"}
154        - Status: Completed
155        """
156
157    # Update session state with summary info
158    run_context.session_state["session_summarized"] = True
159    run_context.session_state["total_plans_summarized"] = len(plans)
160
161    return StepOutput(content=summary.strip())
162
163
164# Define steps using different executor types
165
166research_step = Step(
167    name="Research Step",
168    team=research_team,
169)
170
171content_planning_step = Step(
172    name="Content Planning Step",
173    executor=custom_content_planning_function,
174)
175
176content_summary_step = Step(
177    name="Content Summary Step",
178    executor=content_summary_function,
179)
180
181
182# Define and use examples
183if __name__ == "__main__":
184    content_creation_workflow = Workflow(
185        name="Content Creation Workflow",
186        description="Automated content creation with custom execution options and session state",
187        db=SqliteDb(
188            session_table="workflow_session",
189            db_file="tmp/workflow.db",
190        ),
191        # Define the sequence of steps
192        # First run the research_step, then the content_planning_step, then the summary_step
193        # You can mix and match agents, teams, and even regular python functions directly as steps
194        steps=[research_step, content_planning_step, content_summary_step],
195        # Initialize session state with empty content plans
196        session_state={"content_plans": [], "plan_counter": 0},
197    )
198
199    print("=== First Workflow Run ===")
200    content_creation_workflow.print_response(
201        input="AI trends in 2024",
202        markdown=True,
203    )
204
205    print(
206        f"\nSession State After First Run: {content_creation_workflow.get_session_state()}"
207    )
208
209    print("\n" + "=" * 60 + "\n")
210
211    print("=== Second Workflow Run (Same Session) ===")
212    content_creation_workflow.print_response(
213        input="Machine Learning automation tools",
214        markdown=True,
215    )
216
217    print(f"\nFinal Session State: {content_creation_workflow.get_session_state()}")

1uv pip install -U kern-ai openai yfinance

1python access_session_state_in_custom_python_function_step.py