AgentScope【二、nacos集成-Agent】

大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

AgentScope 集成 Nacos 的核心原理，是基于 A2A (Agent-to-Agent) 协议，将 Nacos 作为统一注册中心，实现 Agent、Skill、MCP、Tool 四类资源的服务注册、健康监测、动态发现与远程RPC调用，从而构建分布式、可扩展的生产级多智能体系统。

1. 整体架构与核心角色

A2A 协议：智能体通信标准，定义统一的 AgentCard（能力元数据）、消息格式与调用规范。
Nacos Registry：充当服务目录与配置中心，存储所有资源的元数据与网络地址。
AgentScope Runtime：服务暴露层，将本地资源包装为 A2A 服务，启动自动注册。
服务发现与调用：Consumer 通过 Nacos 发现 Provider，经 A2A 协议完成跨进程/跨机器调用。

2. 四类资源的注册发现机制

(1) Agent 注册与发现（核心）

注册：Agent 启动 → Runtime 生成 AgentCard（名称、描述、版本、Skills、地址）→ 自动注册至 Nacos。
发现：通过 AgentCardResolver（如 NacosAgentCardResolver）按名称/能力筛选，获取可用实例列表。
健康：Nacos 心跳与健康检查，自动剔除故障节点。

(2) Skill 注册与发现（能力单元）

注册：用 @register_skill 装饰器 → 框架收集元数据 → 随 Agent 一并注册到 Nacos。
发现：从 AgentCard 中读取 Skills 列表，或通过 NacosSkillRepository 全局检索。

(3) MCP (Model Context Protocol) 注册与发现

注册：MCP Server 作为特殊 Agent，将自身服务信息（地址、支持的 Tools）注册到 Nacos。
发现：客户端从 Nacos 获取 MCP 地址 → 建立连接 → 调用 tools/list 获取工具集。

(4) Tool 注册与发现（原子能力）

注册：Tool 隶属于 Skill/MCP，其 Schema 随父节点元数据一同存入 Nacos。
发现：通过 Skill 或 MCP 服务间接发现，调用时框架自动路由。

3. 远程调用原理

寻址：Consumer 从 Nacos 获取 Provider 的 IP:Port。
协议：基于 HTTP/gRPC 封装 A2A 请求（含 Agent/Skill ID、参数、上下文）。
调用：Provider 接收请求 → 分发至对应 Agent/Skill/Tool → 执行逻辑 → 返回结果。
透明化：框架屏蔽网络细节，远程调用与本地调用代码一致。

1. 环境准备

pip install agentscope nacos-sdk-python

2. 步骤 1：启动 Nacos 服务

下载启动 Nacos（默认端口 8848）。
控制台：http://localhost:8848/nacos

3. 步骤 2：Provider 端（注册 Agent & Skill）

from agentscope.agents import AgentBase from agentscope.a2a import NacosRegistry, AgentCardWithRuntimeConfig from agentscope.runtime import AgentApp, DeploymentMode from agentscope.decorators import register_skill from agentscope.memory import InMemoryMemory

— 1. 定义技能 Skill —

@register_skill(

skill_id="weather_query", name="天气查询", description="根据城市名查询实时天气", input_modes=["text"], output_modes=["text"]

) def weather_query(city: str) -> str:

"""查询天气的具体逻辑""" return f"【{city}】今日晴，25°C，空气质量优"

— 2. 定义智能体 Agent —

class WeatherAgent(AgentBase):

def __init__(self): super().__init__( name="WeatherServiceAgent", sys_prompt="你是专业天气助手", memory=InMemoryMemory() ) # 注册技能 self.register_skill(weather_query) def reply(self, x: dict) -> dict: return self.invoke_skill("weather_query", x)

— 3. 配置并注册到 Nacos —

if name == “main”:

# 初始化 Nacos 注册中心 nacos_registry = NacosRegistry( server_addr="127.0.0.1:8848", namespace="agentscope_prod", username="nacos", password="nacos" ) # 创建 Agent 应用 agent_app = AgentApp( app_name="WeatherService", app_description="企业级天气查询智能体服务", a2a_config=AgentCardWithRuntimeConfig( host="0.0.0.0", port=9527, registry=nacos_registry, agent_card={ "version": "1.0.0", "tags": ["weather", "service"] } ) ) # 绑定 Agent 并启动（自动注册） agent_app.bind_agent(WeatherAgent()) # 守护进程启动 agent_app.deploy(mode=DeploymentMode.DAEMON_THREAD)

验证：Nacos 控制台 → 服务列表 → 可见 WeatherService。

4. 步骤 3：Consumer 端（远程发现与调用）

from agentscope.a2a import NacosAgentCardResolver, A2AAgent from agentscope.service import AgentServiceCenter

— 1. 初始化服务中心（连接 Nacos）—

service_center = AgentServiceCenter(

registry=NacosRegistry( server_addr="127.0.0.1:8848", namespace="agentscope_prod" )

)

— 2. 远程发现 Agent —

方式1：按服务名发现

remote_agent = A2AAgent.from_service_name(

service_name="WeatherService", resolver=NacosAgentCardResolver("127.0.0.1:8848")

)

— 3. 远程调用（与本地调用语法一致）—

方式A：直接调用Agent

resp1 = remote_agent({“text”: “北京”}) print(“Agent 调用结果:”, resp1)

方式B：直接调用Skill（推荐）

from agentscope import invoke_skill resp2 = invoke_skill(

skill_name="weather_query", agent_name="WeatherService", input_data={"city": "上海"}

) print(“Skill 调用结果:”, resp2)