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 Model Context Protocol (MCP) tool from the Azure Agents service using a synchronous client.
11 To learn more about Model Context Protocol, visit https://modelcontextprotocol.io/
12
13USAGE:
14 python sample_agents_mcp.py
15
16 Before running the sample:
17
18 pip install azure-ai-projects azure-ai-agents>=1.2.0b3 azure-identity --pre
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 3) MCP_SERVER_URL - The URL of your MCP server endpoint.
26 4) MCP_SERVER_LABEL - A label for your MCP server.
27"""
28
29import os, time
30from azure.ai.projects import AIProjectClient
31from azure.identity import DefaultAzureCredential
32from azure.ai.agents.models import (
33 ListSortOrder,
34 McpTool,
35 RequiredMcpToolCall,
36 RunStepActivityDetails,
37 SubmitToolApprovalAction,
38 ToolApproval,
39)
40
41# Get MCP server configuration from environment variables
42mcp_server_url = os.environ.get("MCP_SERVER_URL", "https://gitmcp.io/Azure/azure-rest-api-specs")
43mcp_server_label = os.environ.get("MCP_SERVER_LABEL", "github")
44
45project_client = AIProjectClient(
46 endpoint=os.environ["PROJECT_ENDPOINT"],
47 credential=DefaultAzureCredential(),
48)
49
50# [START create_agent_with_mcp_tool]
51# Initialize agent MCP tool
52mcp_tool = McpTool(
53 server_label=mcp_server_label,
54 server_url=mcp_server_url,
55 allowed_tools=[], # Optional: specify allowed tools
56)
57
58# You can also add or remove allowed tools dynamically
59search_api_code = "search_azure_rest_api_code"
60mcp_tool.allow_tool(search_api_code)
61print(f"Allowed tools: {mcp_tool.allowed_tools}")
62
63# Create agent with MCP tool and process agent run
64with project_client:
65 agents_client = project_client.agents
66
67 # Create a new agent.
68 # NOTE: To reuse existing agent, fetch it with get_agent(agent_id)
69 agent = agents_client.create_agent(
70 model=os.environ["MODEL_DEPLOYMENT_NAME"],
71 name="my-mcp-agent",
72 instructions="You are a helpful agent that can use MCP tools to assist users. Use the available MCP tools to answer questions and perform tasks.",
73 tools=mcp_tool.definitions,
74 )
75 # [END create_agent_with_mcp_tool]
76
77 print(f"Created agent, ID: {agent.id}")
78 print(f"MCP Server: {mcp_tool.server_label} at {mcp_tool.server_url}")
79
80 # Create thread for communication
81 thread = agents_client.threads.create()
82 print(f"Created thread, ID: {thread.id}")
83
84 # Create message to thread
85 message = agents_client.messages.create(
86 thread_id=thread.id,
87 role="user",
88 content="Please summarize the Azure REST API specifications Readme",
89 )
90 print(f"Created message, ID: {message.id}")
91
92 # [START handle_tool_approvals]
93 # Create and process agent run in thread with MCP tools
94 mcp_tool.update_headers("SuperSecret", "123456")
95 # mcp_tool.set_approval_mode("never") # Uncomment to disable approval requirement
96 run = agents_client.runs.create(thread_id=thread.id, agent_id=agent.id, tool_resources=mcp_tool.resources)
97 print(f"Created run, ID: {run.id}")
98
99 while run.status in ["queued", "in_progress", "requires_action"]:
100 time.sleep(1)
101 run = agents_client.runs.get(thread_id=thread.id, run_id=run.id)
102
103 if run.status == "requires_action" and isinstance(run.required_action, SubmitToolApprovalAction):
104 tool_calls = run.required_action.submit_tool_approval.tool_calls
105 if not tool_calls:
106 print("No tool calls provided - cancelling run")
107 agents_client.runs.cancel(thread_id=thread.id, run_id=run.id)
108 break
109
110 tool_approvals = []
111 for tool_call in tool_calls:
112 if isinstance(tool_call, RequiredMcpToolCall):
113 try:
114 print(f"Approving tool call: {tool_call}")
115 tool_approvals.append(
116 ToolApproval(
117 tool_call_id=tool_call.id,
118 approve=True,
119 headers=mcp_tool.headers,
120 )
121 )
122 except Exception as e:
123 print(f"Error approving tool_call {tool_call.id}: {e}")
124
125 print(f"tool_approvals: {tool_approvals}")
126 if tool_approvals:
127 agents_client.runs.submit_tool_outputs(
128 thread_id=thread.id, run_id=run.id, tool_approvals=tool_approvals
129 )
130
131 print(f"Current run status: {run.status}")
132 # [END handle_tool_approvals]
133
134 print(f"Run completed with status: {run.status}")
135 if run.status == "failed":
136 print(f"Run failed: {run.last_error}")
137
138 # Display run steps and tool calls
139 run_steps = agents_client.run_steps.list(thread_id=thread.id, run_id=run.id)
140
141 # Loop through each step
142 for step in run_steps:
143 print(f"Step {step['id']} status: {step['status']}")
144
145 # Check if there are tool calls in the step details
146 step_details = step.get("step_details", {})
147 tool_calls = step_details.get("tool_calls", [])
148
149 if tool_calls:
150 print(" MCP Tool calls:")
151 for call in tool_calls:
152 print(f" Tool Call ID: {call.get('id')}")
153 print(f" Type: {call.get('type')}")
154
155 if isinstance(step_details, RunStepActivityDetails):
156 for activity in step_details.activities:
157 for function_name, function_definition in activity.tools.items():
158 print(
159 f' The function {function_name} with description "{function_definition.description}" will be called.:'
160 )
161 if len(function_definition.parameters) > 0:
162 print(" Function parameters:")
163 for argument, func_argument in function_definition.parameters.properties.items():
164 print(f" {argument}")
165 print(f" Type: {func_argument.type}")
166 print(f" Description: {func_argument.description}")
167 else:
168 print("This function has no parameters")
169
170 print() # add an extra newline between steps
171
172 # Fetch and log all messages
173 messages = agents_client.messages.list(thread_id=thread.id, order=ListSortOrder.ASCENDING)
174 print("\nConversation:")
175 print("-" * 50)
176 for msg in messages:
177 if msg.text_messages:
178 last_text = msg.text_messages[-1]
179 print(f"{msg.role.upper()}: {last_text.text.value}")
180 print("-" * 50)
181
182 # Example of dynamic tool management
183 print(f"\nDemonstrating dynamic tool management:")
184 print(f"Current allowed tools: {mcp_tool.allowed_tools}")
185
186 # Remove a tool
187 try:
188 mcp_tool.disallow_tool(search_api_code)
189 print(f"After removing {search_api_code}: {mcp_tool.allowed_tools}")
190 except ValueError as e:
191 print(f"Error removing tool: {e}")
192
193 # Clean-up and delete the agent once the run is finished.
194 # NOTE: Comment out this line if you plan to reuse the agent later.
195 agents_client.delete_agent(agent.id)
196 print("Deleted agent")
