STARWARCLOUD STARWAR.AI
返回市场
铸造代理服务远程MCPPython

铸造代理服务远程MCPPython

作者：Azure-Samples9 星标更新：2025-07-07
安装使用来源：spider==github.com

项目介绍

快速开始使用代理服务和远程MCP服务器（Python）

这是一个快速入门模板，用于轻松运行Azure AI Foundry代理服务客户端，并通过Azure Functions远程MCP添加自定义远程MCP服务器到云端。您可以在本地机器上进行克隆/恢复/运行并调试，然后使用azd up在几分钟内将其部署到云端。MCP服务器设计上使用密钥和HTTPS进行安全保护，并允许通过内置认证和/或API管理以及VNET网络隔离来提供更多OAuth选项。

如果您希望查看其他语言版本的示例，请参阅.NET/C#和Node.js/TypeScript版本。

在GitHub Codespaces中打开

以下是使用Azure Functions与Foundry代理服务的远程MCP服务器架构图：

架构图

先决条件

将远程MCP服务器部署到Azure

运行此azd命令以提供函数应用，包括任何所需的Azure资源（如AI Foundry代理服务），并部署您的代码：

azd up

注意：在部署过程中，您将被提示指定一个agentLocation。这必须是AI Foundry支持的区域之一：westus、westus2、uaenorth、southindia或switzerlandnorth。该位置专门用于AI资源（AI服务、搜索、Cosmos DB），可以与您的主要部署位置不同。

此外，还可以使用API管理来提高MCP服务器的安全性和策略，并使用App Service内置身份验证设置您喜欢的OAuth提供商，包括Entra。

从客户端连接到您的远程MCP服务器函数应用

您的客户端需要一个密钥才能调用新的托管SSE端点，其形式为https://<funcappname>.azurewebsites.net/runtime/webhooks/mcp/sse。托管函数默认需要一个系统密钥，可以从门户或CLI (az functionapp keys list --resource-group <resource_group> --name <function_app_name>)获取。获取名为mcp_extension的系统密钥。

Foundry代理服务客户端

  1. 在新终端窗口中切换到代理文件夹：

    cd agent

  2. 根据提供的示例创建一个.env文件。复制.env.example文件：

    copy .env.example .env

  3. 使用已部署的函数应用详情编辑.env文件：

    # Azure AI项目配置
PROJECT_ENDPOINT=https://your-agent-service-resource.services.ai.azure.com/api/projects/your-project-name
MODEL_DEPLOYMENT_NAME=gpt-4.1-mini
MCP_SERVER_LABEL=Azure_Functions_MCP_Server
MCP_SERVER_URL=https://<your-funcappname>.azurewebsites.net/runtime/webhooks/mcp/sse
USER_MESSAGE=创建一个名为snippet1的代码片段，用Python打印'Hello, World!'。

# 必需：Azure Functions扩展密钥用于MCP服务器认证
MCP_EXTENSION_KEY=在这里填写您的mcp_extension系统密钥

    注意：使用azd up部署输出替换以下值：

    • PROJECT_ENDPOINT：您的Azure AI项目端点（来自azd部署输出）
    • <your-funcappname>：您的函数应用名称（来自azd部署输出）
    • 在这里填写您的mcp_extension系统密钥：从Azure门户或CLI获取的mcp_extension系统密钥

  4. 安装代理的Python依赖项：

    pip install -r requirements.txt

  5. 运行代理服务：

    python main.py

    代理将连接到您的远程MCP服务器，并执行USER_MESSAGE环境变量中指定的消息，展示Azure AI Foundry与您部署的MCP服务器之间的集成。

在MCP Inspector中连接到远程MCP服务器

对于MCP Inspector，您可以在URL中包含密钥：

https://<funcappname>.azurewebsites.net/runtime/webhooks/mcp/sse?code=<your-mcp-extension-system-key>

重新部署您的代码

您可以多次运行azd up命令以提供Azure资源并部署代码更新到您的函数应用。

[!注意] 已部署的代码文件总是会被最新的部署包覆盖。

清理资源

当您完成对函数应用及相关资源的工作后，可以使用此命令从Azure删除函数应用及其相关资源，避免产生进一步的成本：

azd down

有用的Azure命令

一旦您的应用程序部署完毕，您可以使用这些命令来管理和监控您的应用程序：

# 从环境文件获取您的函数应用名称
FUNCTION_APP_NAME=$(cat .azure/$(cat .azure/config.json | jq -r '.defaultEnvironment')/env.json | jq -r '.FUNCTION_APP_NAME')
echo $FUNCTION_APP_NAME

# 获取资源组
RESOURCE_GROUP=$(cat .azure/$(cat .azure/config.json | jq -r '.defaultEnvironment')/env.json | jq -r '.AZURE_RESOURCE_GROUP')
echo $RESOURCE_GROUP

# 查看函数应用日志
az webapp log tail --name $FUNCTION_APP_NAME --resource-group $RESOURCE_GROUP

# 重新部署应用程序而不提供新的资源
azd deploy

本地调试MCP服务器函数

由于我们将保存和获取代码片段到Blob存储，因此这个特定示例需要Azure存储模拟器。

  1. 启动Azurite

    docker run -p 10000:10000 -p 10001:10001 -p 10002:10002 \
    mcr.microsoft.com/azure-storage/azurite

注意 如果您使用的是VS Code扩展中的Azurite，现在需要运行Azurite: Start，否则您会看到错误。

从终端本地运行您的MCP服务器

  1. 在新终端窗口中切换到src/mcp_server文件夹：

    cd src/mcp_server

  2. 安装Python依赖项：

    pip install -r requirements.txt

注意 创建虚拟环境是一个最佳实践，以避免依赖问题/冲突，或者如果您正在CodeSpaces中运行。更多信息请参见VS Code中的Python环境。

  1. 本地启动Functions主机：

    func start

注意 默认情况下，这将使用webhooks路由：/runtime/webhooks/mcp/sse。稍后我们将在Azure中使用此路由设置客户端/主机调用的密钥：/runtime/webhooks/mcp/sse?code=<system_key>。

从客户端/主机连接到本地MCP服务器

Foundry代理服务客户端

Foundry代理服务是一个云服务，期望MCP工具也位于云中（例如，相同的VNET或公共互联网）。请继续进行部署Azure以实现远程MCP的步骤。

MCP Inspector

  1. 在新终端窗口中安装并运行MCP Inspector

    npx @modelcontextprotocol/inspector

  2. 按CTRL点击以从应用程序显示的URL加载MCP Inspector Web应用（例如http://0.0.0.0:5173/#resources）

  3. 设置传输类型为SSE

  4. 设置URL为您运行的函数应用的SSE端点并连接：

    http://0.0.0.0:7071/runtime/webhooks/mcp/sse

注意 此步骤在CodeSpaces中无法工作。请继续进行部署到远程MCP。

源代码

get_snippet和save_snippet端点的函数代码定义在src/mcp_server目录下的Python文件中。MCP函数注解将这些函数暴露为MCP服务器工具。

下面是function_app.py文件中的实际代码：


@app.generic_trigger(arg_name="context", type="mcpToolTrigger", toolName="hello", 
                     description="Hello world.", 
                     toolProperties="[]")
def hello_mcp(context) -> None:
    """
    返回问候消息的简单函数。

    参数：
        context: 触发上下文（在此函数中未使用）。

    返回：
        str: 问候消息。
    """
    return "Hello I am MCPTool!"


@app.generic_trigger(
    arg_name="context",
    type="mcpToolTrigger",
    toolName="getsnippet",
    description="按名称检索代码片段。",
    toolProperties=tool_properties_get_snippets_json
)
@app.generic_input_binding(
    arg_name="file",
    type="blob",
    connection="AzureWebJobsStorage",
    path=_BLOB_PATH
)
def get_snippet(file: func.InputStream, context) -> str:
    """
    从Azure Blob存储中按名称检索代码片段。

    参数：
        file (func.InputStream): 读取Azure Blob存储中代码片段的输入绑定。
        context: 包含输入参数的触发上下文。

    返回：
        str: 代码片段的内容或错误消息。
    """
    snippet_content = file.read().decode("utf-8")
    logging.info(f"检索到的代码片段：{snippet_content}")
    return snippet_content


@app.generic_trigger(
    arg_name="context",
    type="mcpToolTrigger",
    toolName="savesnippet",
    description="按名称保存代码片段。",
    toolProperties=tool_properties_save_snippets_json
)                   
@app.generic_output_binding(
    arg_name="file",
    type="blob",
    connection="AzureWebJobsStorage",
    path=_BLOB_PATH
)
def save_snippet(file: func.Out[str], context) -> str:
    content = json.loads(context)
    snippet_name_from_args = content["arguments"][_SNIPPET_NAME_PROPERTY_NAME]
    snippet_content_from_args = content["arguments"][_SNIPPET_PROPERTY_NAME]

    if not snippet_name_from_args:
        return "未提供代码片段名称"

    if not snippet_content_from_args:
        return "未提供代码片段内容"
 
    file.set(snippet_content_from_args)
    logging.info(f"保存的代码片段：{snippet_content_from_args}")
    return f"代码片段'{snippet_content_from_args}'保存成功"

请注意，host.json文件还包括对实验性捆绑包的引用，这是使用此功能的应用程序所必需的：

"extensionBundle": {
  "id": "Microsoft.Azure.Functions.ExtensionBundle.Experimental",
  "version": "[4.*, 5.0.0)"
}

下一步