技術速遞｜使用 Semantic Kernel 與 A2A 協議構建多智能體解決方案

作者：盧建暉 - 微軟高級云技術布道師

翻譯/排版：Alan Wang

在快速發展的 AI 應用開發領域，能夠協調多個智能體已成為構建復雜企業級解決方案的關鍵。雖然單個 AI 智能體擅長特定任務，但復雜的業務場景往往需要跨平臺、跨框架甚至跨組織邊界的專業智能體協同工作。這正是微軟的 Semantic Kernel 編排功能與 A2A（Agent-to-Agent）協議相結合，構建真正互操作多智能體系統的重要基礎。

理解 A2A 協議：超越傳統工具集成

A2A 協議由 Google 于 2025 年 4 月推出，并得到了超過 50 家技術合作伙伴的支持，旨在解決 AI 生態系統中的一個核心難題：讓智能體能夠以對等的身份進行溝通與協作，而不僅僅是作為工具被調用。與側重連接智能體與外部工具和數據源的 MCP（模型上下文協議）不同，A2A 構建了一個專門為智能體間交互設計的標準化通信層。

A2A 的核心能力

通過 Agent 卡片發現智能體：每個支持 A2A 的智能體都會公開一個機器可讀的 “Agent 卡片”，這是一個 JSON 文檔，用于展示其能力、接入端點、支持的消息類型、認證要求及運行時元數據。這種發現機制使客戶端智能體能夠動態識別并選擇最適合執行特定任務的遠程智能體。
通過生命周期跟蹤進行任務管理：A2A 的所有交互都圍繞有著明確生命周期的獨立任務展開，這些任務可以即時完成，也可以持續較長時間，并提供實時狀態更新，使該協議既適用于快速的 API 調用，也適用于可能耗時數小時或數天的復雜研究操作。
豐富的消息交換：通信通過結構化消息進行，消息包含具有特定內容類型的“部分”。這使得智能體能夠協商合適的交互方式，并交換多種類型的數據，包括文本、結構化 JSON、文件，甚至多媒體流。
企業級安全保障：A2A 構建在常見的 Web 標準之上（如 HTTP、JSON-RPC 2.0、Server-Sent Events），并結合了企業級的身份驗證與權限控制，支持 OpenAPI 的認證機制，在不暴露智能體內部狀態或專有工具的前提下，實現安全的協作。

A2A 與 MCP：互為補充，而非競爭關系

一個常見的誤解是 A2A 與 Anthropic 的模型上下文協議（MCP）存在競爭關系。實際上，這兩個協議服務于智能體 AI 技術棧中的不同層面：

MCP 連接智能體與工具和數據 —— 使其能夠訪問外部 API、數據庫、文件系統等結構化資源
A2A 連接智能體與其他智能體 —— 實現對等協作、任務委派和分布式問題解決

你可以把它理解為：MCP 是給智能體一把錘子，而 A2A 是教它如何與一個施工隊協同工作。大多數復雜的應用程序都會同時利用這兩種協議。

Semantic Kernel：編排引擎

微軟的 Semantic Kernel 為構建支持 A2A 的多智能體系統提供了理想的基礎。作為一個開源 SDK，Semantic Kernel 在以下方面表現出色：

插件化架構：通過可復用插件輕松擴展智能體能力
多模型支持：協調不同的 AI 模型以執行專門任務
企業集成：無縫連接現有的企業系統與 API
智能體框架：提供實驗性但強大的多智能體協同能力

為什么要結合 Semantic Kernel 與 A2A？

框架無關互操作性：Semantic Kernel 智能體可與 LangGraph、CrewAI、Google ADK 或其他 A2A 兼容框架智能體進行通信
保留 Semantic Kernel 的優勢：在享受 Semantic Kernel 插件生態與提示工程、企業能力的同時，實現跨平臺兼容
漸進遷移路徑：現有 Semantic Kernel 應用可逐步采用 A2A，無需大規模重構
云原生設計：支持認證、日志、可觀測性，滿足企業場景需求

A2A 支持系統的架構模式

在使用 Semantic Kernel 和 A2A 設計多智能體系統時，我們的實現架構展示了幾個關鍵模式：

使用 Azure AI Foundry 的集中路由

我們采用的主要模式是使用由 Azure AI Foundry 驅動的中央路由智能體，智能地將任務委派給各個專業的遠程智能體：
在這里插入圖片描述

關鍵組件

主機智能體：基于 Azure AI Agents 的中央路由系統，負責智能決策
A2A 協議：標準化的智能體間通信機制
Semantic Kernel：集成 MCP 的高級智能體框架
遠程智能體：使用不同通信協議的專業任務執行者

優勢

通過 Azure AI Foundry 的線程實現集中式對話狀態管理
基于智能體能力和用戶意圖進行智能任務委派
在不同智能體交互中提供一致的用戶體驗
明確的審計記錄和完善的錯誤處理機制

多協議智能體通信

系統展示了不同通信協議如何實現共存：
在這里插入圖片描述

通信模式：

A2A HTTP/JSON-RPC：適用于具備標準化發現能力的通用工具型智能體
STDIO：用于基于進程的智能體，如 Playwright 自動化
Server-Sent Events（SSE）：用于 Azure 中的無服務器 MCP 函數

MCP 與 A2A 的混合集成

我們的架構展示了 MCP 協議與 A2A 協議如何相互補充：
在這里插入圖片描述

面向工具的 MCP：直接集成 Azure Functions、開發工具和數據源
面向智能體的 A2A：實現智能體間通信與任務委派
Semantic Kernel：作為編排層，橋接兩種協議

實施詳解

項目結構與組件

我們的多智能體系統采用模塊化架構：
在這里插入圖片描述

開發環境設置

要構建這個支持 A2A 的 Semantic Kernel 系統，我們需要：

# Initialize Python project with uv (recommended)
uv init multi_agent_system
cd multi_agent_system
# Core dependencies for Semantic Kernel and Azure integration
uv add semantic-kernel[azure]
uv add azure-identity
uv add azure-ai-agents
uv add python-dotenv
# A2A protocol dependencies
uv add a2a-client
uv add httpx
# MCP integration dependencies
uv add semantic-kernel[mcp]
# Web interface dependencies
uv add gradio
# Development dependencies
uv add --dev pytest pytest-asyncio

環境配置

配置以下環境變量：

# Azure AI Foundry configuration
AZURE_AI_AGENT_ENDPOINT=https://your-ai-foundry-endpoint.azure.com
AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME=Your AI Foundry Model Deployment Name
# Remote agent endpoints
PLAYWRIGHT_AGENT_URL=http://localhost:10001
TOOL_AGENT_URL=http://localhost:10002
# Optional: MCP server configuration
MCP_SSE_URL=http://localhost:7071/runtime/webhooks/mcp/sse

創建中央路由智能體

我們系統的核心是利用 Azure AI Foundry 實現智能任務委派的路由智能體：

import json
import os
import time
import uuid
from typing import Any, Dict, List
import httpx
from a2a.client import A2ACardResolver
from azure.ai.agents import AgentsClient
from azure.identity import DefaultAzureCredential
from dotenv import load_dotenv
class RoutingAgent:"""Central routing agent powered by Azure AI Foundry."""def __init__(self):self.remote_agent_connections = {}self.cards = {}self.agents_client = AgentsClient(endpoint=os.environ["AZURE_AI_AGENT_ENDPOINT"],credential=DefaultAzureCredential(),)self.azure_agent = Noneself.current_thread = Noneasync def initialize(self, remote_agent_addresses: list[str]):"""Initialize with A2A agent discovery."""# Discover remote agents via A2A protocolasync with httpx.AsyncClient(timeout=30) as client:for address in remote_agent_addresses:try:card_resolver= A2ACardResolver(client, address)card = await card_resolver.get_agent_card()from remote_agent_connection import RemoteAgentConnectionsremote_connection = RemoteAgentConnections(agent_card=card, agent_url=address)self.remote_agent_connections[card.name] = remote_connectionself.cards[card.name] = cardexcept Exception as e:print(f'Failed to connect to agent at {address}: {e}')# Create Azure AI agent for intelligent routingawait self._create_azure_agent()async def _create_azure_agent(self):"""Create Azure AI agent with function calling capabilities."""instructions = self._get_routing_instructions()# Define function for task delegationtools = [{"type": "function","function": {"name": "send_message","description": "Delegate task to specialized remote agent","parameters": {"type": "object","properties": {"agent_name": {"type": "string"},"task": {"type": "string"}},"required": ["agent_name", "task"]}}}]model_name = os.environ.get("AZURE_AI_AGENT_MODEL_DEPLOYMENT_NAME", "gpt-4")self.azure_agent = self.agents_client.create_agent(model=model_name,name="routing-agent",instructions=instructions,tools=tools)self.current_thread = self.agents_client.threads.create()print(f"Routing agent initialized: {self.azure_agent.id}")def _get_routing_instructions(self) -> str:"""Generate context-aware routing instructions."""agent_info = [{'name': card.name, 'description': card.description}for card in self.cards.values()]return f"""You are an intelligent routing agent for a multi-agent system.
Available Specialist Agents:
{json.dumps(agent_info, indent=2)}

路由指引：

網頁自動化、截圖、瀏覽器任務 → Playwright 智能體
開發任務、文件操作、代碼庫管理 → 工具智能體
委派任務時務必提供完整的任務上下文
向用戶說明哪個智能體正在處理他們的請求

構建集成 MCP 的專業智能體

我們的遠程智能體利用 Semantic Kernel 的 MCP 集成，實現可擴展能力：

from semantic_kernel.agents import AzureAIAgent, AzureAIAgentSettings
from semantic_kernel.connectors.mcp import MCPStdioPlugin, MCPSsePlugin
from azure.identity.aio import DefaultAzureCredential
class SemanticKernelMCPAgent:"""Specialized agent with MCP plugin integration."""def __init__(self):self.agent = Noneself.client = Noneself.credential = Noneself.plugins = []async def initialize_playwright_agent(self):"""Initialize with Playwright automation via MCP STDIO."""try:self.credential= DefaultAzureCredential()self.client = await AzureAIAgent.create_client(credential=self.credential).__aenter__()# Create Playwright MCP pluginplaywright_plugin = MCPStdioPlugin(name="Playwright",command="npx",args=["@playwright/mcp@latest"],)await playwright_plugin.__aenter__()self.plugins.append(playwright_plugin)# Create specialized agentagent_definition = await self.client.agents.create_agent(model=AzureAIAgentSettings().model_deployment_name,name="PlaywrightAgent",instructions=("You are a web automation specialist. Use Playwright to ""navigate websites, take screenshots, interact with elements, ""and perform browser automation tasks."),)self.agent = AzureAIAgent(client=self.client,definition=agent_definition,plugins=self.plugins,)except Exception as e:await self.cleanup()raiseasync def initialize_tools_agent(self, mcp_url: str):"""Initialize with development tools via MCP SSE."""try:self.credential= DefaultAzureCredential()self.client = AzureAIAgent.create_client(credential=self.credential)# Create development tools MCP plugintools_plugin = MCPSsePlugin(name="DevTools",url=mcp_url,)await tools_plugin.__aenter__()self.plugins.append(tools_plugin)agent_definition = await self.client.agents.create_agent(model=AzureAIAgentSettings().model_deployment_name,name="DevAssistant",instructions=("You are a development assistant. Help with repository ""management, file operations, opening projects in VS Code, ""and other development tasks."),)self.agent = AzureAIAgent(client=self.client,definition=agent_definition,plugins=self.plugins,)except Exception as e:await self.cleanup()raiseasync def invoke(self, user_input: str) -> dict[str, Any]:"""Process tasks through the specialized agent."""if not self.agent:return{'is_task_complete': False,'content': 'Agent not initialized.',}try:responses = []async for response in self.agent.invoke(messages=user_input,thread=self.thread,):responses.append(str(response))self.thread = response.threadreturn {'is_task_complete': True,'content': "\n".join(responses) or "No response received.",}except Exception as e:return {'is_task_complete': False,'content': f'Error: {str(e)}',}

基于 Gradio 的 Web 界面

系統提供了由 Gradio 驅動的現代聊天界面：

import asyncio
import gradio as gr
from routing_agent import RoutingAgent
async def get_response_from_agent(message: str, history: list[gr.ChatMessage]
) -> gr.ChatMessage:"""Process user messages through the routing system."""global ROUTING_AGENTtry:response= 
awaitROUTING_AGENT.process_user_message(message)return gr.ChatMessage(role="assistant", content=response)except Exception as e:return gr.ChatMessage(role="assistant",content=f"? Error: {str(e)}")
async def main():"""Launch the multi-agent system."""# Initialize routing agentglobal ROUTING_AGENTROUTING_AGENT= 
awaitRoutingAgent.create([os.getenv('PLAYWRIGHT_AGENT_URL', 'http://localhost:10001'),os.getenv('TOOL_AGENT_URL', 'http://localhost:10002'),])ROUTING_AGENT.create_agent()# Create Gradio interfacewith gr.Blocks(theme=gr.themes.Ocean()) as demo:gr.Markdown("# 🤖 Azure AI Multi-Agent System")gr.ChatInterface(get_response_from_agent,title="Chat with AI Agents",examples=["Navigate to github.com/microsoft and take a screenshot","Clone repository https://github.com/microsoft/semantic-kernel","Open the cloned project in VS Code",])demo.launch(server_name="0.0.0.0", server_port=8083)

部署與運維

本地開發環境：

# Start MCP server (Azure Functions)
cd mcp_sse_server/MCPAzureFunc
func start
?
# Start remote agents in separate terminals
cd remote_agents/playwright_agent && uv run .
cd remote_agents/tool_agent && uv run .
?
# Start the host agent with web interface
cd host_agent && uv run .

生產環境注意事項：

將每個智能體作為獨立微服務部署，推薦使用 Azure Container Apps
利用 Azure Service Bus 實現可靠的智能體發現與通信
使用 Azure Application Insights 實施全面的日志記錄
配置合適的身份驗證和網絡安全措施

實際應用示例

基于我們的實現，以下是多智能體系統處理不同類型請求的實用示例：

示例一：網頁自動化任務

User: "Navigate to github.com/microsoft and take a screenshot"
Flow:
1. Host Agent (Azure AI) analyzes the request
2. Identifies this as a web automation task  
3. Delegates to Playwright Agent via A2A protocol
4. Playwright Agent uses MCP STDIO to execute browser automation
5. Returns screenshot and navigation details to user

示例二：開發工作流

User: "Clone https://github.com/microsoft/semantic-kernel and open it in VS Code"
Flow:
1. Host Agent recognizes repository management + IDE operation
2. Delegates to Tool Agent via A2A protocol  
3. Tool Agent uses MCP SSE connection to Azure Functions
4. Executes git clone and VS Code launch commands
5. Reports success status back to user