通过 Python 开始使用 Azure MCP 服务器

Azure MCP 服务器使用模型上下文协议（MCP）来标准化 AI 应用与外部工具和数据源之间的集成，使 AI 系统能够执行对上下文敏感的操作，并且能够识别和处理您的 Azure 资源。

本文介绍如何完成以下任务：

• 安装和向 Azure MCP 服务器进行身份验证
• 使用自定义 Python 客户端连接到 Azure MCP 服务器
• 运行提示以测试 Azure MCP 服务器操作并管理 Azure 资源

先决条件

• 具有活动订阅的 Azure 帐户
• 已安装 Python 3.9 或更高版本
• Node.js 已安装 LTS

用于本地开发的登录

Azure MCP 服务器通过 Microsoft Entra ID 使用基于令牌的身份验证提供无缝身份验证体验。 在内部，Azure MCP 服务器使用 DefaultAzureCredentialAzure 标识库 对用户进行身份验证。

需要使用 Azure 帐户登录到本地支持 DefaultAzureCredential 的工具之一，才能使用 Azure MCP 服务器。 使用终端窗口登录，例如 Visual Studio Code 终端：

az login

Connect-AzAccount

azd auth login

成功登录到上述工具之一后，Azure MCP 服务器可以自动发现您的凭据，并使用这些凭据对 Azure 服务进行身份验证和执行操作。

创建 Python 应用

完成以下步骤以创建 Python 应用。 该应用连接到 AI 模型，充当连接到 Azure MCP 服务器的 MCP 客户端的主机。

创建项目

1. 在所选编辑器中打开一个空文件夹。

2. 创建一个名为requirements.txt的新文件，并添加以下库依赖项：

   mcp
   azure-identity
   openai
   logging
   

3. 在同一文件夹中，创建一 .env 个名为并添加以下环境变量的新文件：

   AZURE_OPENAI_ENDPOINT=<your-azure-openai-endpoint>
   AZURE_OPENAI_MODEL=<your-model-deployment-name>
   

4. 创建一个名为 main.py 用于保存应用代码的空文件。

创建环境和安装依赖项

1. 在新文件夹中打开终端，并为应用创建 Python 虚拟环境：

   python -m venv venv
   

2. 激活虚拟环境：

   venv\Scripts\activate
   

3. 从 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 客户端。
• 初始化 MCP 客户端，以使用标准 I/O 传输与 Azure MCP 服务器交互。
• 从 MCP 服务器检索和显示可用工具的列表。
• 实现聊天循环以处理用户提示、利用工具和处理工具调用。

运行并测试应用

完成以下步骤以测试 .NET 主机应用：

1. 在打开到项目的根目录中的终端窗口中，运行以下命令以启动应用：

   python main.py
   

2. 应用运行后，请输入以下测试提示：

   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!
   

3. 使用其他相关提示来浏览和测试 Azure MCP 操作，例如：

   List all of the storage accounts in my subscription
   Get the available tables in my storage accounts
   

后续步骤