McpToolRegistrationService 类

为 Azure Foundry 代理提供 MCP 工具注册服务。

此服务使用 Azure AI SDK 通过 Azure Foundry 代理处理 MCP(模型上下文协议)工具服务器的注册和管理。 它提供 MCP 服务器与 Azure Foundry 代理框架之间的无缝集成。

功能:

  • 自动 MCP 服务器发现和配置
  • Azure 标识与 DefaultAzureCredential 的集成
  • 工具定义和资源管理
  • 支持开发(ToolingManifest.json)和生产(网关 API)方案
  • 全面的错误处理和日志记录

构造函数

McpToolRegistrationService()

参数

名称 说明
logger
默认值: None
credential
默认值: None

示例


>>> service = McpToolRegistrationService()
>>> service.add_tool_servers_to_agent(project_client, agent_id, token)

方法

__init__

初始化 Azure Foundry 的 MCP 工具注册服务。

__new__
add_tool_servers_to_agent

将 MCP 工具服务器添加到 Azure Foundry 代理。

send_chat_history

从 Azure AI Foundry 检索聊天历史记录并将其发送到 MCP 平台。

此方法使用提供的客户端和线程 ID 从 Azure AI Foundry 代理 API 检索消息,将其转换为 ChatHistoryMessage 格式,并将其发送到 MCP 平台。

send_chat_history_messages

将 Azure AI Foundry 聊天历史记录消息发送到 MCP 平台。

此方法接受一系列 Azure AI Foundry ThreadMessage 对象,将其转换为 ChatHistoryMessage 格式,并将其发送到 MCP 平台以进行实时威胁防护。

__init__

初始化 Azure Foundry 的 MCP 工具注册服务。

__init__(logger: Logger | None = None, credential: DefaultAzureCredential | None = None)

参数

名称 说明
logger

日志记录作的记录器实例。

默认值: None
credential
<xref:azure.identity._credentials.default.DefaultAzureCredential> | None

用于身份验证的 Azure 凭据。 如果为 None,将使用 DefaultAzureCredential。

默认值: None

__new__

__new__(**kwargs)

add_tool_servers_to_agent

将 MCP 工具服务器添加到 Azure Foundry 代理。

async add_tool_servers_to_agent(project_client: AIProjectClient, auth: Authorization, auth_handler_name: str, context: TurnContext, auth_token: str | None = None) -> None

参数

名称 说明
project_client
必需
<xref:azure.ai.projects._patch.AIProjectClient>

Azure Foundry AIProjectClient 实例。

auth
必需

令牌交换的授权处理程序。

auth_handler_name
必需
str

授权处理程序的名称。

context
必需

为当前作轮次上下文。

auth_token
str | None

用于访问 MCP 服务器的身份验证令牌。

默认值: None

返回

类型 说明

例外

类型 说明

如果project_client为 None 或必需参数无效。

如果在 MCP 工具注册期间出错。

send_chat_history

从 Azure AI Foundry 检索聊天历史记录并将其发送到 MCP 平台。

此方法使用提供的客户端和线程 ID 从 Azure AI Foundry 代理 API 检索消息,将其转换为 ChatHistoryMessage 格式,并将其发送到 MCP 平台。

async send_chat_history(agents_client: AgentsClient, thread_id: str, turn_context: TurnContext, tool_options: ToolOptions | None = None) -> OperationResult

参数

名称 说明
agents_client
必需
<xref:azure.ai.agents._patch.AgentsClient>

Azure AI Foundry AgentsClient 实例。

thread_id
必需
str

包含要发送的消息的线程 ID。

turn_context
必需

来自包含聊天信息的代理 SDK 的 TurnContext。

tool_options

请求的可选配置。

默认值: None

返回

类型 说明

指示成功或失败的 OperationResult。

例外

类型 说明

如果agents_client、thread_id或turn_context为 None/empty。

示例


>>> from azure.ai.agents import AgentsClient
>>> from azure.identity import DefaultAzureCredential
>>>
>>> client = AgentsClient(endpoint, credential=DefaultAzureCredential())
>>> service = McpToolRegistrationService()
>>> result = await service.send_chat_history(
...     client, thread_id, turn_context
... )

send_chat_history_messages

将 Azure AI Foundry 聊天历史记录消息发送到 MCP 平台。

此方法接受一系列 Azure AI Foundry ThreadMessage 对象,将其转换为 ChatHistoryMessage 格式,并将其发送到 MCP 平台以进行实时威胁防护。

async send_chat_history_messages(turn_context: TurnContext, messages: Sequence[ThreadMessage], tool_options: ToolOptions | None = None) -> OperationResult

参数

名称 说明
turn_context
必需

来自包含聊天信息的代理 SDK 的 TurnContext。

messages
必需
Sequence[<xref:azure.ai.agents.models._patch.ThreadMessage>]

要发送的 Azure AI Foundry ThreadMessage 对象的序列。

tool_options

请求的可选配置。

默认值: None

返回

类型 说明

指示成功或失败的 OperationResult。

例外

类型 说明

如果turn_context或消息为 None。

示例


>>> service = McpToolRegistrationService()
>>> messages = await agents_client.messages.list(thread_id=thread_id)
>>> result = await service.send_chat_history_messages(
...     turn_context, list(messages)
... )
>>> if result.succeeded:
...     print("Chat history sent successfully")