Workflow with Input Schema Validation

This example demonstrates **Workflows** support for input schema validation using Pydantic models to ensure type safety and data integrity at the workflow entry point.

This example shows how to use input schema validation in workflows to enforce type safety and data structure validation. By defining an input_schema with a Pydantic model, you can ensure that your workflow receives properly structured and validated data before execution begins.

Key Features:

Type Safety: Automatic validation of input data against Pydantic models
Structure Validation: Ensure all required fields are present and correctly typed
Clear Contracts: Define exactly what data your workflow expects
Error Prevention: Catch invalid inputs before workflow execution begins
Multiple Input Formats: Support for Pydantic models and matching dictionaries

1from typing import List
2
3from kern.agent import Agent
4from kern.db.sqlite import SqliteDb
5from kern.models.openai import OpenAIResponses
6from kern.team import Team
7from kern.tools.hackernews import HackerNewsTools
8from kern.tools.yfinance import YFinanceTools
9from kern.workflow.step import Step
10from kern.workflow.workflow import Workflow
11from pydantic import BaseModel, Field
12
13
14class DifferentModel(BaseModel):
15    name: str
16
17
18class ResearchTopic(BaseModel):
19    """Structured research topic with specific requirements"""
20
21    topic: str
22    focus_areas: List[str] = Field(description="Specific areas to focus on")
23    target_audience: str = Field(description="Who this research is for")
24    sources_required: int = Field(description="Number of sources needed", default=5)
25
26
27# Define agents
28hackernews_agent = Agent(
29    name="Hackernews Agent",
30    model=OpenAIResponses(id="gpt-5.2"),
31    tools=[HackerNewsTools()],
32    role="Extract key insights and content from Hackernews posts",
33)
34finance_agent = Agent(
35    name="Finance Agent",
36    model=OpenAIResponses(id="gpt-5.2"),
37    tools=[YFinanceTools()],
38    role="Get stock prices and financial data",
39)
40
41# Define research team for complex analysis
42research_team = Team(
43    name="Research Team",
44    members=[hackernews_agent, finance_agent],
45    instructions="Research tech topics from Hackernews and the web",
46)
47
48content_planner = Agent(
49    name="Content Planner",
50    model=OpenAIResponses(id="gpt-5.2"),
51    instructions=[
52        "Plan a content schedule over 4 weeks for the provided topic and research content",
53        "Ensure that I have posts for 3 posts per week",
54    ],
55)
56
57# Define steps
58research_step = Step(
59    name="Research Step",
60    team=research_team,
61)
62
63content_planning_step = Step(
64    name="Content Planning Step",
65    agent=content_planner,
66)
67
68# Create and use workflow
69if __name__ == "__main__":
70    content_creation_workflow = Workflow(
71        name="Content Creation Workflow",
72        description="Automated content creation from blog posts to social media",
73        db=SqliteDb(
74            session_table="workflow_session",
75            db_file="tmp/workflow.db",
76        ),
77        steps=[research_step, content_planning_step],
78        input_schema=ResearchTopic,  # <-- Define input schema for validation
79    )
80
81    print("=== Example 1: Valid Pydantic Model Input ===")
82    research_topic = ResearchTopic(
83        topic="AI trends in 2024",
84        focus_areas=[
85            "Machine Learning",
86            "Natural Language Processing",
87            "Computer Vision",
88            "AI Ethics",
89        ],
90        target_audience="Tech professionals and business leaders",
91    )
92
93    # This will work properly - input matches the schema
94    content_creation_workflow.print_response(
95        input=research_topic,
96        markdown=True,
97    )
98
99    print("\n=== Example 2: Valid Dictionary Input ===")
100    # This will also work - dict matches ResearchTopic structure
101    content_creation_workflow.print_response(
102        input={
103            "topic": "AI trends in 2024",
104            "focus_areas": ["Machine Learning", "Computer Vision"],
105            "target_audience": "Tech professionals",
106            "sources_required": 8
107        },
108        markdown=True,
109    )
110
111    print("\n=== Example 3: Missing Required Fields (Commented - Would Fail) ===")
112    # This would fail - missing required 'target_audience' field
113    # Uncomment to see validation error:
114    # content_creation_workflow.print_response(
115    #     input=ResearchTopic(
116    #         topic="AI trends in 2024",
117    #         focus_areas=[
118    #             "Machine Learning",
119    #             "Natural Language Processing",
120    #             "Computer Vision",
121    #             "AI Ethics",
122    #         ],
123    #         # target_audience missing - will raise ValidationError
124    #     ),
125    #     markdown=True,
126    # )
127
128    print("\n=== Example 4: Wrong Model Type (Commented - Would Fail) ===")
129    # This would fail - different Pydantic model provided
130    # Uncomment to see validation error:
131    # content_creation_workflow.print_response(
132    #     input=DifferentModel(name="test"),
133    #     markdown=True,
134    # )
135
136    print("\n=== Example 5: Type Mismatch (Commented - Would Fail) ===")
137    # This would fail - wrong data types
138    # Uncomment to see validation error:
139    # content_creation_workflow.print_response(
140    #     input={
141    #         "topic": 123,  # Should be string
142    #         "focus_areas": "Machine Learning",  # Should be List[str]
143    #         "target_audience": ["audience1", "audience2"],  # Should be string
144    #     },
145    #     markdown=True,
146    # )