Step with Function using Additional Data

1from kern.agent import Agent
2from kern.db.sqlite import SqliteDb
3from kern.models.openai import OpenAIResponses
4from kern.team import Team
5from kern.tools.hackernews import HackerNewsTools
6from kern.tools.yfinance import YFinanceTools
7from kern.workflow.step import Step, StepInput, StepOutput
8from kern.workflow.workflow import Workflow
9
10# Define agents
11hackernews_agent = Agent(
12    name="Hackernews Agent",
13    model=OpenAIResponses(id="gpt-5.2"),
14    tools=[HackerNewsTools()],
15    instructions="Extract key insights and content from Hackernews posts",
16)
17
18finance_agent = Agent(
19    name="Finance Agent",
20    model=OpenAIResponses(id="gpt-5.2"),
21    tools=[YFinanceTools()],
22    instructions="Get stock prices and financial data",
23)
24
25# Define research team for complex analysis
26research_team = Team(
27    name="Research Team",
28    members=[hackernews_agent, finance_agent],
29    instructions="Analyze content and create comprehensive social media strategy",
30)
31
32content_planner = Agent(
33    name="Content Planner",
34    model=OpenAIResponses(id="gpt-5.2"),
35    instructions=[
36        "Plan a content schedule over 4 weeks for the provided topic and research content",
37        "Ensure that I have posts for 3 posts per week",
38    ],
39)
40
41
42def custom_content_planning_function(step_input: StepInput) -> StepOutput:
43    """
44    Custom function that does intelligent content planning with context awareness
45    Now also uses additional_data for extra context
46    """
47    message = step_input.input
48    previous_step_content = step_input.previous_step_content
49
50    # Access additional_data that was passed with the workflow
51    additional_data = step_input.additional_data or {}
52    user_email = additional_data.get("user_email", "No email provided")
53    priority = additional_data.get("priority", "normal")
54    client_type = additional_data.get("client_type", "standard")
55
56    # Create intelligent planning prompt
57    planning_prompt = f"""
58        STRATEGIC CONTENT PLANNING REQUEST:
59
60        Core Topic: {message}
61
62        Research Results: {previous_step_content[:500] if previous_step_content else "No research results"}
63
64        Additional Context:
65        - Client Type: {client_type}
66        - Priority Level: {priority}
67        - Contact Email: {user_email}
68
69        Planning Requirements:
70        1. Create a comprehensive content strategy based on the research
71        2. Leverage the research findings effectively
72        3. Identify content formats and channels
73        4. Provide timeline and priority recommendations
74        5. Include engagement and distribution strategies
75        {"6. Mark as HIGH PRIORITY delivery" if priority == "high" else "6. Standard delivery timeline"}
76
77        Please create a detailed, actionable content plan.
78    """
79
80    try:
81        response = content_planner.run(planning_prompt)
82
83        enhanced_content = f"""
84            ## Strategic Content Plan
85
86            **Planning Topic:** {message}
87
88            **Client Details:**
89            - Type: {client_type}
90            - Priority: {priority.upper()}
91            - Contact: {user_email}
92
93            **Research Integration:** {"Research-based" if previous_step_content else "No research foundation"}
94
95            **Content Strategy:**
96            {response.content}
97
98            **Custom Planning Enhancements:**
99            - Research Integration: {"High" if previous_step_content else "Baseline"}
100            - Strategic Alignment: Optimized for multi-channel distribution
101            - Execution Ready: Detailed action items included
102            - Priority Level: {priority.upper()}
103        """.strip()
104
105        return StepOutput(content=enhanced_content, response=response)
106
107    except Exception as e:
108        return StepOutput(
109            content=f"Custom content planning failed: {str(e)}",
110            success=False,
111        )
112
113
114# Define steps using different executor types
115
116research_step = Step(
117    name="Research Step",
118    team=research_team,
119)
120
121content_planning_step = Step(
122    name="Content Planning Step",
123    executor=custom_content_planning_function,
124)
125
126
127# Define and use examples
128if __name__ == "__main__":
129    content_creation_workflow = Workflow(
130        name="Content Creation Workflow",
131        description="Automated content creation with custom execution options",
132        db=SqliteDb(
133            session_table="workflow_session",
134            db_file="tmp/workflow.db",
135        ),
136        steps=[research_step, content_planning_step],
137    )
138
139    # Run workflow with additional_data
140    content_creation_workflow.print_response(
141        input="AI trends in 2024",
142        additional_data={
143            "user_email": "michael@dundermifflin.com",
144            "priority": "high",
145            "client_type": "enterprise",
146        },
147        markdown=True,
148        stream=True,
149    )
150
151    print("\n" + "=" * 60 + "\n")