Implementing Ticket Schemas

# 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).",
    )

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

# 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()}

# 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],
)

(.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]: