Coding guide to create an Agentic Audic Asychronous Tichetant Assistant using Pydantai agents, Pydantic V2, and SQLITE DATHABASE

!pip install --upgrade pip
!pip install pydantic-ai

import os
from getpass import getpass


if "GEMINI_API_KEY" not in os.environ:
    os.environ["GEMINI_API_KEY"] = getpass("Enter your Google Gemini API key: ")

!pip install nest_asyncio

import sqlite3
import uuid
from dataclasses import dataclass
from typing import Literal


from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext

conn = sqlite3.connect(":memory:")
conn.execute("""
CREATE TABLE tickets (
    ticket_id TEXT PRIMARY KEY,
    summary   TEXT NOT NULL,
    severity  TEXT NOT NULL,
    department TEXT NOT NULL,
    status    TEXT NOT NULL
)
""")
conn.commit()

@dataclass
class TicketingDependencies:
    """Carries our DB connection into system prompts and tools."""
    db: sqlite3.Connection


class CreateTicketOutput(BaseModel):
    ticket_id: str = Field(..., description="Unique ticket identifier")
    summary: str   = Field(..., description="Text summary of the issue")
    severity: Literal["low","medium","high"] = Field(..., description="Urgency level")
    department: str = Field(..., description="Responsible department")
    status: Literal["open"] = Field("open", description="Initial ticket status")


class TicketStatusOutput(BaseModel):
    ticket_id: str = Field(..., description="Unique ticket identifier")
    status: Literal["open","in_progress","resolved"] = Field(..., description="Current ticket status")

create_agent = Agent(
    "google-gla:gemini-2.0-flash",
    deps_type=TicketingDependencies,
    output_type=CreateTicketOutput,
    system_prompt="You are a ticketing assistant. Use the `create_ticket` tool to log new issues."
)


@create_agent.tool
async def create_ticket(
    ctx: RunContext[TicketingDependencies],
    summary: str,
    severity: Literal["low","medium","high"],
    department: str
) -> CreateTicketOutput:
    """
    Logs a new ticket in the database.
    """
    tid = str(uuid.uuid4())
    ctx.deps.db.execute(
        "INSERT INTO tickets VALUES (?,?,?,?,?)",
        (tid, summary, severity, department, "open")
    )
    ctx.deps.db.commit()
    return CreateTicketOutput(
        ticket_id=tid,
        summary=summary,
        severity=severity,
        department=department,
        status="open"
    )

status_agent = Agent(
    "google-gla:gemini-2.0-flash",
    deps_type=TicketingDependencies,
    output_type=TicketStatusOutput,
    system_prompt="You are a ticketing assistant. Use the `get_ticket_status` tool to retrieve current status."
)


@status_agent.tool
async def get_ticket_status(
    ctx: RunContext[TicketingDependencies],
    ticket_id: str
) -> TicketStatusOutput:
    """
    Fetches the ticket status from the database.
    """
    cur = ctx.deps.db.execute(
        "SELECT status FROM tickets WHERE ticket_id = ?", (ticket_id,)
    )
    row = cur.fetchone()
    if not row:
        raise ValueError(f"No ticket found for ID {ticket_id!r}")
    return TicketStatusOutput(ticket_id=ticket_id, status=row[0])

deps = TicketingDependencies(db=conn)


create_result = await create_agent.run(
    "My printer on 3rd floor shows a paper jam error.", deps=deps
)


print("Created Ticket →")
print(create_result.output.model_dump_json(indent=2))


tid = create_result.output.ticket_id
status_result = await status_agent.run(
    f"What's the status of ticket {tid}?", deps=deps
)


print("Ticket Status →")
print(status_result.output.model_dump_json(indent=2))