Azure MCP 服务器使用模型上下文协议(MCP)来标准化 AI 应用与外部工具和数据源之间的集成,使 AI 系统能够执行对上下文敏感的操作,并且能够识别和处理您的 Azure 资源。
本文介绍如何完成以下任务:
- 安装和向 Azure MCP 服务器进行身份验证
- 使用自定义 Python 客户端连接到 Azure MCP 服务器
- 运行提示以测试 Azure MCP 服务器操作并管理 Azure 资源
先决条件
- 具有活动订阅的 Azure 帐户
- 已安装 Python 3.9 或更高版本
- Node.js 已安装 LTS
注释
要通过 Azure MCP 服务器访问的 Azure 资源必须已存在于 Azure 订阅中。 此外,用户帐户必须具有为这些资源分配的必要 RBAC 角色和权限 。
登录到 Azure MCP 服务器进行本地开发
Azure MCP 服务器使用适用于 .NET 的 Azure 标识库对 Microsoft Entra ID 进行身份验证。 服务器支持 两种身份验证模式:
-
代理模式:使用操作系统的原生身份验证(如 Windows Web 帐户管理器)与
InteractiveBrowserCredential一起使用。 - 凭据链模式:按顺序尝试多种身份验证方法:环境变量、Visual Studio Code、Visual Studio、Azure CLI、Azure PowerShell、Azure 开发人员 CLI 和交互式浏览器身份验证。
使用以下任一方法登录:
- 打开命令面板(
Ctrl+Shift+P或在 Mac 上使用Cmd+Shift+P)。 - 运行 Azure:登录 并按照提示进行作。
登录后,Azure MCP 服务器可以根据权限对 Azure 服务进行身份验证和运行作。
创建 Python 应用
完成以下步骤以创建 Python 应用(主机应用)。 该应用连接到 AI 模型,充当连接到 Azure MCP 服务器(运行 MCP 协议的本地进程)的 MCP 客户端的主机。
创建项目
在所选编辑器中打开一个空文件夹。
创建一个名为
requirements.txt的新文件,并添加以下库依赖项:mcp azure-identity openai logging在同一文件夹中,创建一
.env个名为并添加以下环境变量的新文件:AZURE_OPENAI_ENDPOINT=<your-azure-openai-endpoint> AZURE_OPENAI_MODEL=<your-model-deployment-name>创建一个名为
main.py用于保存应用代码的空文件。
创建环境和安装依赖项
在新文件夹中打开终端,并为应用创建 Python 虚拟环境:
python -m venv venv激活虚拟环境:
venv\Scripts\activate从
requirements.txt安装依赖项。pip install -r requirements.txt
添加应用代码
将 main.py 的内容替换为以下代码:
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import AzureOpenAI
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client
import json, os, logging, asyncio
from dotenv import load_dotenv
# Setup logging and load environment variables
logger = logging.getLogger(__name__)
load_dotenv()
# Azure OpenAI configuration
AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
AZURE_OPENAI_MODEL = os.getenv("AZURE_OPENAI_MODEL", "gpt-4o")
# Initialize Azure credentials
token_provider = get_bearer_token_provider(
DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
)
async def run():
# Initialize Azure OpenAI client
client = AzureOpenAI(
azure_endpoint=AZURE_OPENAI_ENDPOINT,
api_version="2024-04-01-preview",
azure_ad_token_provider=token_provider
)
# MCP client configurations
server_params = StdioServerParameters(
command="npx",
args=["-y", "@azure/mcp@latest", "server", "start"],
env=None
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# List available tools
tools = await session.list_tools()
for tool in tools.tools: print(tool.name)
# Format tools for Azure OpenAI
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
} for tool in tools.tools]
# Start conversational loop
messages = []
while True:
try:
user_input = input("\nPrompt: ")
messages.append({"role": "user", "content": user_input})
# First API call with tool configuration
response = client.chat.completions.create(
model = AZURE_OPENAI_MODEL,
messages = messages,
tools = available_tools)
# Process the model's response
response_message = response.choices[0].message
messages.append(response_message)
# Handle function calls
if response_message.tool_calls:
for tool_call in response_message.tool_calls:
function_args = json.loads(tool_call.function.arguments)
result = await session.call_tool(tool_call.function.name, function_args)
# Add the tool response to the messages
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"name": tool_call.function.name,
"content": result.content,
})
else:
logger.info("No tool calls were made by the model")
# Get the final response from the model
final_response = client.chat.completions.create(
model = AZURE_OPENAI_MODEL,
messages = messages,
tools = available_tools)
for item in final_response.choices:
print(item.message.content)
except Exception as e:
logger.error(f"Error in conversation loop: {e}")
print(f"An error occurred: {e}")
if __name__ == "__main__":
import asyncio
asyncio.run(run())
前面的代码完成以下任务:
- 设置日志记录并从文件加载环境变量
.env。 - 使用
azure-identity和openai库配置 Azure OpenAI 客户端。 - 使用标准 I/O 传输初始化 MCP 客户端以与 Azure MCP 服务器(本地进程)交互。
- 从 Azure MCP 服务器检索并显示可用工具列表(MCP 注册的 Azure 操作)。
- 实现聊天循环以处理用户提示、利用工具和处理工具调用。
配置参数:
| 参数 | Description | Example |
|---|---|---|
AZURE_OPENAI_ENDPOINT |
Azure OpenAI 服务终结点 | https://your-resource.openai.azure.com/ |
AZURE_OPENAI_MODEL |
模型部署名称 | gpt-4o |
| 令牌范围 | Azure 认知服务 OAuth 范围 | https://cognitiveservices.azure.com/.default |
| 身份验证 | 使用 DefaultAzureCredential (Azure CLI、托管身份或其他凭据链) |
请参阅 Azure 标识文档 |
| 所需的 RBAC | Azure OpenAI 资源上的认知服务用户角色或等效角色 | 通过 Azure 门户或 CLI 分配 |
运行并测试应用
完成以下步骤以测试 Python 应用:
在打开到项目的根目录中的终端窗口中,运行以下命令以启动应用:
python main.py成功验证:应用应显示可用 Azure MCP 服务器工具的列表,然后显示输入
Prompt:。应用运行后,请输入以下测试提示:
List all of the resource groups in my subscription上一提示的输出应类似于以下文本:
The following resource groups are available for your subscription: 1. **DefaultResourceGroup-EUS** (Location: `eastus`) 2. **rg-testing** (Location: `centralus`) 3. **rg-azd** (Location: `eastus2`) 4. **msdocs-sample** (Location: `southcentralus`) 14. **ai-testing** (Location: `eastus2`) Let me know if you need further details or actions related to any of these resource groups!使用其他相关提示来浏览和测试 Azure MCP 操作,例如:
List all of the storage accounts in my subscription Get the available tables in my storage accounts