Async MySQL for Team

1docker run -d \
2  --name mysql \
3  -e MYSQL_ROOT_PASSWORD=ai \
4  -e MYSQL_DATABASE=ai \
5  -e MYSQL_USER=ai \
6  -e MYSQL_PASSWORD=ai \
7  -p 3306:3306 \
8  -d mysql:8

1import asyncio
2import uuid
3from typing import List
4
5from kern.agent import Agent
6from kern.db.base import SessionType
7from kern.db.mysql import AsyncMySQLDb
8from kern.team import Team
9from kern.tools.hackernews import HackerNewsTools
10from kern.tools.hackernews import HackerNewsTools
11from pydantic import BaseModel
12
13db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
14db = AsyncMySQLDb(db_url=db_url)
15
16class Article(BaseModel):
17    title: str
18    summary: str
19    reference_links: List[str]
20
21hn_researcher = Agent(
22    name="HackerNews Researcher",
23    role="Gets top stories from hackernews.",
24    tools=[HackerNewsTools()],
25)
26
27web_searcher = Agent(
28    name="Web Searcher",
29    role="Searches the web for information on a topic",
30    tools=[HackerNewsTools()],
31    add_datetime_to_context=True,
32)
33
34hn_team = Team(
35    name="HackerNews Team",
36    members=[hn_researcher, web_searcher],
37    db=db,
38    instructions=[
39        "First, search hackernews for what the user is asking about.",
40        "Then, ask the web searcher to search for each story to get more information.",
41        "Finally, provide a thoughtful and engaging summary.",
42    ],
43    output_schema=Article,
44    markdown=True,
45    show_members_responses=True,
46)
47
48async def main():
49    """Run the agent queries in the same event loop"""
50    session_id = str(uuid.uuid4())
51    await hn_team.aprint_response(
52        "Write an article about the top 2 stories on hackernews", session_id=session_id
53    )
54    session_data = await db.get_session(
55        session_id=session_id, session_type=SessionType.TEAM
56    )
57    print("\n=== SESSION DATA ===")
58    print(session_data.to_dict())
59
60if __name__ == "__main__":
61    asyncio.run(main())