Build an AI-Powered Asynchronous Ticketing Assistant with PydanticAI, Pydantic v2, and SQLite

Overview of the AI-Powered Ticketing Assistant

This tutorial walks you through building a fully functional ticketing assistant powered by Agentic AI using the PydanticAI library. The assistant operates asynchronously, handling ticket creation and status checking through natural language prompts powered by Google Gemini.

Setting Up the Environment

To begin, you update pip and install PydanticAI, which enables the creation of type-safe AI agents integrated with Pydantic models and large language models (LLMs).

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

You then securely provide your Google Gemini API key for authentication:

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: ")

Additionally, you install nest_asyncio to handle asynchronous event loops smoothly in environments like Colab:

!pip install nest_asyncio

Database and Data Models

An in-memory SQLite database is created to store ticket data with fields for ticket ID, summary, severity, department, and status.

import sqlite3
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()

Two Pydantic v2 models define the schema for ticket creation and status output, ensuring type safety and validation.

from pydantic import BaseModel, Field
from typing import Literal
 
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")

Building the Agents

Two PydanticAI agents are created: one for ticket creation and one for checking ticket status. Each agent uses the Google Gemini model.

Create Ticket Agent

from pydantic_ai import Agent, RunContext
import uuid
 
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:
    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"
    )

Ticket Status Agent

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:
    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])

Running the Agents

You package the database connection into a dependencies object and invoke the agents using natural language prompts.

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))

This workflow demonstrates how Agentic AI combined with PydanticAI automates ticket management with robust data validation and a lightweight database backend. The system can be extended with additional agents, AI models, or real-world API integrations while maintaining structured and reliable data handling.