agent.py

import asyncio
import json
import os
import dotenv
from typing import Annotated

from azure.identity.aio import DefaultAzureCredential, AzureCliCredential

from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings
from semantic_kernel.functions import kernel_function
from pizza_api_client import get_pizzas, get_toppings, create_order, cancel_order, get_order_status, get_orders

"""
The following sample demonstrates how to create an Azure AI agent that answers
questions about a sample menu using a Semantic Kernel Plugin.
"""

# Define a sample plugin for the sample
class MenuPlugin:
    """A Menu Plugin that uses the pizza API client."""

    @kernel_function(description="Returns the full pizza menu as JSON.")
    def get_specials(self) -> Annotated[str, "Returns the available pizzas from the menu."]:
        pizzas = get_pizzas()
        return json.dumps(pizzas)

    @kernel_function(description="Returns the full toppings list as JSON.")
    def get_toppings(self) -> Annotated[str, "Returns the available toppings from the menu."]:
        toppings = get_toppings()
        return json.dumps(toppings)

    @kernel_function(description="Create a new pizza order for the user. The 'items' parameter must be a JSON string in the following format: [ { 'pizzaId': 'pizza-1', 'quantity': 1, 'extraToppingIds': ['topping-1', 'topping-2'] } ]")
    def create_order(self, items: Annotated[str, "Order items as a JSON string. Format: [ { 'pizzaId': 'pizza-1', 'quantity': 1, 'extraToppingIds': ['topping-1', 'topping-2'] } ]"]) -> Annotated[str, "Returns the created order as JSON."]:
        # items should be a JSON string representing a list of order items in the specified format
        items_list = json.loads(items)
        order = create_order(items_list)
        return json.dumps(order)

    @kernel_function(description="Cancel an existing order for the user.")
    def cancel_order(self, order_id: Annotated[str, "Order ID to cancel."]) -> Annotated[str, "Returns the cancellation result as JSON."]:
        result = cancel_order(order_id)
        return json.dumps(result)

    @kernel_function(description="Get the status of an existing order.")
    def get_order_status(self, order_id: Annotated[str, "Order ID to check."]) -> Annotated[str, "Returns the order status as JSON."]:
        status = get_order_status(order_id)
        return json.dumps(status)

    @kernel_function(description="Get all orders, optionally filtered by status, or last (e.g., '60m', '2h').")
    def get_orders(self, status: Annotated[str, "Status to filter orders (comma-separated, e.g. 'pending,ready')."], last: Annotated[str, "Filter orders created in the last X minutes/hours (e.g. '60m', '2h')."]) -> Annotated[str, "Returns a list of orders as JSON."]:
        # Accept empty string as None for optional params
        status = status or None
        last = last or None
        orders = get_orders(status=status, last=last)
        return json.dumps(orders)


# Simulate a conversation with the agent
USER_INPUTS = [
    "Hello",
    "What pizzas are on the menu?",
#"Can you list all available toppings?",
    "I would like to create an order for 1 Margherita pizza with pepperoni.",
    "can you show all my orders?",
    "What is the status of my latest order?",
    "Please cancel that order.",
    "Thank you",
]

dotenv.load_dotenv()

async def main() -> None:
    tenant_id = os.getenv("AZURE_TENANT_ID")
    async with (
        AzureCliCredential() as creds,
        AzureAIAgent.create_client(credential=creds, endpoint=os.getenv("PROJECT_ENDPOINT")) as client,
    ):
        # 1. Get the agent on the Azure AI agent service
        agent_definition = await client.agents.get_agent(os.getenv("AGENT_ID"))

        # 2. Create a Semantic Kernel agent for the Azure AI agent
        agent = AzureAIAgent(
            client=client,
            definition=agent_definition,
            plugins=[MenuPlugin()],  # Add the plugin to the agent
        )

        # 3. Create a thread for the agent
        # If no thread is provided, a new thread will be
        # created and returned with the initial response
        thread = None

        try:
            for user_input in USER_INPUTS:
                print(f"# User: {user_input}")
                # 4. Invoke the agent for the specified thread for response
                async for response in agent.invoke(
                    messages=user_input,
                    thread=thread,
                ):
                    print(f"# {response.name}: {response}")
                    thread = response.thread
        finally:
            # 5. Cleanup: Delete the thread
            await thread.delete() if thread else None

        """
        Sample Output:
        # "Hello",
        # "What pizzas are on the menu?",
        # ...
        """


if __name__ == "__main__":
    asyncio.run(main())