Stream Hook

1"""
2Stream Hook
3=============================
4
5Example demonstrating sending a notification to the user after an agent generates a response.
6"""
7
8import asyncio
9
10from kern.agent import Agent
11from kern.models.openai import OpenAIResponses
12from kern.run import RunContext
13from kern.run.agent import RunOutput
14from kern.tools.yfinance import YFinanceTools
15
16
17def send_notification(run_output: RunOutput, run_context: RunContext) -> None:
18    """
19    Post-hook: Send a notification to the user.
20    """
21    if run_context.metadata is None:
22        return
23    email = run_context.metadata.get("email")
24    if email:
25        send_email(email, run_output.content)
26
27
28def send_email(email: str, content: str) -> None:
29    """
30    Send an email to the user. Mock, just for the example.
31    """
32    print(f"Sending email to {email}: {content}")
33
34
35# ---------------------------------------------------------------------------
36# Create Agent
37# ---------------------------------------------------------------------------
38async def main():
39    # Agent with comprehensive output validation
40
41    # ---------------------------------------------------------------------------
42    # Create Agent
43    # ---------------------------------------------------------------------------
44
45    agent = Agent(
46        name="Financial Report Agent",
47        model=OpenAIResponses(id="gpt-5-mini"),
48        post_hooks=[send_notification],
49        tools=[YFinanceTools()],
50        instructions=[
51            "You are a helpful financial report agent.",
52            "Generate a financial report for the given company.",
53            "Keep it short and concise.",
54        ],
55    )
56
57    # Run the agent
58    await agent.aprint_response(
59        "Generate a financial report for Apple (AAPL).",
60        user_id="user_123",
61        metadata={"email": "test@example.com"},
62        stream=True,
63    )
64
65
66# ---------------------------------------------------------------------------
67# Run Agent
68# ---------------------------------------------------------------------------
69if __name__ == "__main__":
70    asyncio.run(main())

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