Implementing Ticket Schemas

In this lesson we’ll refactor the single-file demo into a small, well-structured project and implement a typed, validated ticket workflow using Pydantic. The goal is to keep the agent focused on orchestration while tools implement domain logic and validation.

Goals for this lesson:

Create a schemas package with a Pydantic Ticket model.
Move domain logic into a tools package.
Add a typed create_ticket tool that returns a validated, structured ticket.
Keep the agent orchestrator thin—use tools for lookup, checks, and ticket creation.

File layout

Path	Purpose
`schemas/ticket.py`	Pydantic Ticket model defining the canonical ticket shape
`tools/helpdesk_tools.py`	Implementations: user lookup, service status check, and ticket creation tools
`agent.py`	Agent setup and FunctionTool wrappers that call the tools

1) Ticket schema (Pydantic)

Use a Pydantic model to enforce a consistent ticket shape on the Python side. The LLM does not need to produce the exact JSON; instead, the agent will call a tool which builds and validates the Pydantic model, returning a reliable dict. Example schemas/ticket.py:

# schemas/ticket.py
from typing import Literal, Optional
from pydantic import BaseModel, Field
from datetime import datetime

class Ticket(BaseModel):
    """Schema for an IT helpdesk ticket."""
    ticket_id: str = Field(
        description="Human-readable ID for the ticket, e.g. IT-1A2B3C4D."
    )
    summary: str = Field(
        description="Short summary of the user's issue."
    )
    service: str = Field(
        description="The affected service, e.g. 'email', 'vpn', 'gitlab', 'wifi'."
    )
    user_email: str = Field(
        description="The user's work email address related to this ticket."
    )
    severity: Literal["low", "medium", "high"] = Field(
        default="medium",
        description="Severity of the issue based on impact and urgency.",
    )
    status: Literal["open", "in_progress", "resolved"] = Field(
        default="open",
        description="Current status of the ticket in the helpdesk workflow.",
    )
    department: Optional[str] = Field(
        default=None,
        description="User's department, if known.",
    )
    created_at: datetime = Field(
        description="When the ticket was created (UTC).",
    )
    updated_at: Optional[datetime] = Field(
        default=None,
        description="When the ticket was last updated (UTC).",
    )

References:

Pydantic docs: https://docs.pydantic.dev/

2) Typed tool input: CreateTicketArgs

Define a typed input model for the create-ticket tool. This ensures the tool receives validated input and makes intent explicit. Example CreateTicketArgs (place this inside tools/helpdesk_tools.py or a shared module):

from pydantic import BaseModel, Field
from typing import Optional, Literal

class CreateTicketArgs(BaseModel):
    summary: str = Field(description="Concise summary of the issue in the agent's words.")
    description: Optional[str] = Field(default=None, description="Longer description from the user, if provided.")
    service: str = Field(description="Affected service, e.g., 'email', 'vpn', 'gitlab', 'wifi'.")
    user_email: Optional[str] = Field(default=None, description="User's email address, if known.")
    severity: Literal["low", "medium", "high"] = Field(default="medium", description="Impact-based severity.")
    department: Optional[str] = Field(default=None, description="User's department, if known.")

3) Tool implementations (lookup, service status, create ticket)

Move domain logic into tools/helpdesk_tools.py. For the demo we use simple in-memory backends and return structured dicts indicating “status” plus result payload or error message. Example excerpts for tools/helpdesk_tools.py:

# tools/helpdesk_tools.py
from typing import Dict, Any
from datetime import datetime
import uuid

from pydantic import BaseModel
from ..schemas.ticket import Ticket  # relative import when tools/ and schemas/ are sibling packages
from .helpdesk_tools import CreateTicketArgs  # if CreateTicketArgs is in this module, adjust accordingly

# Fake backends (demo)
_FAKE_USER_DIRECTORY: Dict[str, Dict[str, Any]] = {
    "alice@example.com": {"email": "alice@example.com", "name": "Alice", "status": "active", "department": "Engineering"},
    # add more as needed
}

_FAKE_SERVICE_STATUS: Dict[str, str] = {
    "vpn": "degraded",
    "email": "operational",
    "gitlab": "outage",
    "wifi": "operational",
}

def lookup_user_impl(email: str) -> Dict[str, Any]:
    """Look up a user in the internal directory.

    Returns:
        dict: { "status": "success" | "error", "user": {...}, "error_message": "..." }
    """
    if not email:
        return {"status": "error", "error_message": "No email provided."}

    user = _FAKE_USER_DIRECTORY.get(email.lower())
    if not user:
        return {
            "status": "error",
            "error_message": f"No user found for email '{email}'.",
        }
    return {"status": "success", "user": user}

def check_service_status_impl(service_name: str) -> Dict[str, Any]:
    """Check the known fake service status map.

    Returns:
        dict: {
            "status": "success" | "error",
            "service": normalized_name,
            "status_text": "operational" | "degraded" | "outage",
            "error_message": "..."
        }
    """
    if not service_name:
        return {"status": "error", "error_message": "No service name provided."}

    normalized = service_name.strip().lower()
    status = _FAKE_SERVICE_STATUS.get(normalized)
    if not status:
        return {
            "status": "error",
            "error_message": (
                f"Unknown service '{service_name}'. "
                f"Known services: {', '.join(sorted(_FAKE_SERVICE_STATUS.keys()))}."
            ),
        }
    return {
        "status": "success",
        "service": normalized,
        "status_text": status,
    }

def create_ticket_impl(args: CreateTicketArgs) -> Dict[str, Any]:
    """Create a validated Ticket object and return it as a dict."""
    ticket_id = f"IT-{uuid.uuid4().hex[:8].upper()}"
    ticket = Ticket(
        ticket_id=ticket_id,
        summary=args.summary,
        service=args.service.lower(),
        user_email=(args.user_email or "").lower(),
        severity=args.severity,
        status="open",
        department=args.department,
        created_at=datetime.utcnow(),
        updated_at=None,
    )
    # Use model_dump() for Pydantic v2 compatibility; .dict() for v1.
    return {"status": "success", "ticket": ticket.model_dump()}

Notes:

These implementations always return a small structured dict with a status field. This makes it easy for the agent to branch on success vs error.
In a real system, replace fake backends with database calls, API clients, or ticketing system integrations.

4) Wrapping tools for the agent

In your agent module you wrap these functions with FunctionTool so the agent can call them. The agent keeps instruction and orchestration logic, while tools encapsulate domain behavior and validation. Example agent.py excerpt:

# agent.py (excerpt)
from google.adk.tools import FunctionTool
from tools.helpdesk_tools import (
    lookup_user_impl,
    check_service_status_impl,
    create_ticket_impl,
)

lookup_user_tool = FunctionTool(func=lookup_user_impl, name="lookup_user")
check_service_status_tool = FunctionTool(func=check_service_status_impl, name="check_service_status")
create_ticket_tool = FunctionTool(func=create_ticket_impl, name="create_ticket")

root_agent = Agent(
    model='gemini-2.5-flash',
    name="helpdesk_root_agent",
    description=(
        "Smart IT Helpdesk assistant that troubleshoots common IT issues "
        "using clarifying questions and internal tools."
    ),
    instruction=(
        "You are a friendly but efficient IT helpdesk assistant for an internal company.\n"
        "\n"
        "You are running inside a multi-turn session. ADK will provide tools you can call.\n"
        "1) Use tools when it adds value:\n"
        "   - Call 'lookup_user' when you have an email and it would help to know account status.\n"
        "   - Call 'check_service_status' when the issue involves a known service (email, vpn, gitlab, wifi).\n"
        "2) Decide whether to create a ticket:\n"
        "   - Create a ticket if the issue is severe, troubleshooting didn't resolve it, or the user asks.\n"
        "   - If creating a ticket, call 'create_ticket' with the fields: summary, service, user_email, severity.\n"
        "3) Give next steps: concise numbered steps or suggestions based on tool results.\n"
    ),
    tools=[lookup_user_tool, check_service_status_tool, create_ticket_tool],
)

Using tool-level structured outputs (tools return validated dicts produced from Pydantic models) provides strong guarantees about data shape while still letting the LLM decide when and how to call those tools.

5) Example run (console)

A typical session shows the agent calling tools, receiving structured outputs, and creating a ticket:

(.venv) jeremy@MACSTUDIO ticketpro % ADK run helpdesk_agent
Running agent helpdesk_root_agent, type exit to exit.
[user]: I've been completely unable to connect to the VPN all morning, and our whole team is blocked. My email is alice@example.com
[helpdesk_root_agent]: Thanks, Alice. I've checked into this, and it looks like the VPN service is currently degraded. This means it's experiencing issues, which is likely why you and your team are blocked.

I'm creating a high-priority ticket for this so our IT team can investigate further and work on a resolution.

I've opened a ticket for you: IT-811464F0.

Summary: Your team is unable to connect to the VPN because the service is currently degraded.
Service: VPN
Severity: High

Next Steps:
1. Our IT team is aware of the degraded VPN service and is actively working on it.
2. Please monitor company announcements for updates on the VPN status.
3. You can try connecting to the VPN again periodically.
[user]:

Behind the scenes

The agent called check_service_status and received a structured response: {status: "success", service: "vpn", status_text: "degraded"}.
Given the impact, the agent built a CreateTicketArgs payload and invoked create_ticket.
create_ticket_impl instantiated the Ticket Pydantic model, validated it, and returned ticket.model_dump() as a structured result.
The resulting dict can be logged, stored in a database, or sent to a ticketing API.

Summary

Refactored the demo into schemas/ and tools/.
Added a Pydantic Ticket model and a typed create-ticket tool that returns a validated ticket dictionary.
Moved domain logic into tools and left the agent focused on orchestration and instructions.
This pattern—tool-level structured outputs + typed inputs—maintains LLM flexibility while guaranteeing reliable data shapes for downstream systems.

Next steps

Add grounding sources (internal IT policies, runbooks) so the agent consults authoritative documentation.
Integrate a real user directory and ticketing API to replace demo backends.
Add telemetry and audit logging for ticket creation and tool calls.

Links and references

Pydantic: https://docs.pydantic.dev/
Python datetime docs: https://docs.python.org/3/library/datetime.html
Python uuid docs: https://docs.python.org/3/library/uuid.html
(ADK) Consult your ADK provider docs for FunctionTool and agent runtime specifics.

Introduction

ADK Fundamentals

Deploying and Operating ADK Agents

Implementing Ticket Schemas

1) Ticket schema (Pydantic)

2) Typed tool input: CreateTicketArgs

3) Tool implementations (lookup, service status, create ticket)

4) Wrapping tools for the agent

5) Example run (console)

Watch Video

​1) Ticket schema (Pydantic)

​2) Typed tool input: CreateTicketArgs

​3) Tool implementations (lookup, service status, create ticket)

​4) Wrapping tools for the agent

​5) Example run (console)

Watch Video

1) Ticket schema (Pydantic)

2) Typed tool input: CreateTicketArgs

3) Tool implementations (lookup, service status, create ticket)

4) Wrapping tools for the agent

5) Example run (console)