1# pylint: disable=line-too-long,useless-suppression
2# ------------------------------------
3# Copyright (c) Microsoft Corporation.
4# Licensed under the MIT License.
5# ------------------------------------
6 
7"""
8DESCRIPTION:
9    This sample demonstrates how to use agent operations with the
10    OpenAPI tool from the Azure Agents service using a synchronous client.
11    To learn more about OpenAPI specs, visit https://learn.microsoft.com/openapi
12 
13USAGE:
14    python sample_agents_openapi.py
15 
16    Before running the sample:
17 
18    pip install azure-ai-projects azure-ai-agents azure-identity jsonref
19 
20    Set these environment variables with your own values:
21    1) PROJECT_ENDPOINT - The Azure AI Project endpoint, as found in the Overview
22                          page of your Azure AI Foundry portal.
23    2) MODEL_DEPLOYMENT_NAME - The deployment name of the AI model, as found under the "Name" column in
24       the "Models + endpoints" tab in your Azure AI Foundry project.
25"""
26 
27import os
28import jsonref
29from azure.ai.projects import AIProjectClient
30from azure.identity import DefaultAzureCredential
31from azure.ai.agents.models import OpenApiTool, OpenApiAnonymousAuthDetails
32 
33weather_asset_file_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../assets/weather_openapi.json"))
34 
35countries_asset_file_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../assets/countries.json"))
36 
37project_client = AIProjectClient(
38    endpoint=os.environ["PROJECT_ENDPOINT"],
39    credential=DefaultAzureCredential(),
40)
41# [START create_agent_with_openapi]
42 
43with open(weather_asset_file_path, "r") as f:
44    openapi_weather = jsonref.loads(f.read())
45 
46with open(countries_asset_file_path, "r") as f:
47    openapi_countries = jsonref.loads(f.read())
48 
49# Create Auth object for the OpenApiTool (note that connection or managed identity auth setup requires additional setup in Azure)
50auth = OpenApiAnonymousAuthDetails()
51 
52# Initialize agent OpenApi tool using the read in OpenAPI spec
53openapi_tool = OpenApiTool(
54    name="get_weather", spec=openapi_weather, description="Retrieve weather information for a location", auth=auth
55)
56openapi_tool.add_definition(
57    name="get_countries", spec=openapi_countries, description="Retrieve a list of countries", auth=auth
58)
59 
60# Create agent with OpenApi tool and process agent run
61with project_client:
62    agents_client = project_client.agents
63 
64    agent = agents_client.create_agent(
65        model=os.environ["MODEL_DEPLOYMENT_NAME"],
66        name="my-agent",
67        instructions="You are a helpful agent",
68        tools=openapi_tool.definitions,
69    )
70    # [END create_agent_with_openapi]
71 
72    print(f"Created agent, ID: {agent.id}")
73 
74    # Create thread for communication
75    thread = agents_client.threads.create()
76    print(f"Created thread, ID: {thread.id}")
77 
78    # Create message to thread
79    message = agents_client.messages.create(
80        thread_id=thread.id,
81        role="user",
82        content="What's the weather in Seattle and What is the name and population of the country that uses currency with abbreviation THB?",
83    )
84    print(f"Created message, ID: {message.id}")
85 
86    # Create and process agent run in thread with tools
87    run = agents_client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id)
88    print(f"Run finished with status: {run.status}")
89 
90    if run.status == "failed":
91        print(f"Run failed: {run.last_error}")
92 
93    run_steps = agents_client.run_steps.list(thread_id=thread.id, run_id=run.id)
94 
95    # Loop through each step
96    for step in run_steps:
97        print(f"Step {step['id']} status: {step['status']}")
98 
99        # Check if there are tool calls in the step details
100        step_details = step.get("step_details", {})
101        tool_calls = step_details.get("tool_calls", [])
102 
103        if tool_calls:
104            print("  Tool calls:")
105            for call in tool_calls:
106                print(f"    Tool Call ID: {call.get('id')}")
107                print(f"    Type: {call.get('type')}")
108 
109                function_details = call.get("function", {})
110                if function_details:
111                    print(f"    Function name: {function_details.get('name')}")
112        print()  # add an extra newline between steps
113 
114    # Delete the agent when done
115    agents_client.delete_agent(agent.id)
116    print("Deleted agent")
117 
118    # Fetch and log all messages
119    messages = agents_client.messages.list(thread_id=thread.id)
120    for msg in messages:
121        if msg.text_messages:
122            last_text = msg.text_messages[-1]
123            print(f"{msg.role}: {last_text.text.value}")