nacos-sdk-python

Nacos provides Prompt template management, including retrieval, subscription, and variable-based rendering.

Prompt Rendering: Prompts use {{variableName}} placeholders. You can render a prompt using the .render(variables_dict) method. If a variable is not provided in the dictionary, the renderer will automatically use the defaultValue defined in the PromptVariable object within the prompt template.

Key Operations:

• Get Prompt: Retrieve a prompt via get_prompt using GetPromptParam (requires prompt_key, optional version and label).
• Subscribe: Listen for prompt changes via subscribe_prompt using SubscribePromptParam (requires prompt_key and subscribe_callback).
• Unsubscribe: Stop listening via unsubscribe_prompt.

# Example: Get and Render Prompt
from v2.nacos.ai.model.ai_param import GetPromptParam

prompt = await ai_client.get_prompt(
    GetPromptParam(prompt_key='my-prompt', version='1.0.0')
)

# Render with variables
result = prompt.render({"name": "Alice", "place": "Nacos"})
print(result)  # e.g. "Hello Alice, welcome to Nacos!"

# Example: Subscribe to Prompt changes
from v2.nacos.ai.model.ai_param import SubscribePromptParam

async def prompt_listener(prompt_key, prompt):
    print(f"Prompt changed: {prompt_key}, version: {prompt.version}")

prompt = await ai_client.subscribe_prompt(
    SubscribePromptParam(
        prompt_key='my-prompt',
        version='1.0.0',
        subscribe_callback=prompt_listener
    )
)

Nacos provides management capabilities for Model Context Protocol (MCP) Servers, including registration, discovery, and subscription.

Key Operations:

• Get MCP Server: Retrieve details using get_mcp_server with GetMcpServerParam (requires mcp_name, optional version).
• Publish MCP Server: Register a new server using release_mcp_server with ReleaseMcpServerParam. Requires server_spec (McpServerBasicInfo).
• Register Endpoint: Map a server to a network location using register_mcp_server_endpoint with RegisterMcpServerEndpointParam (requires mcp_name, address, and port).
• Subscribe: Listen for changes using subscribe_mcp_server with SubscribeMcpServerParam. Requires a subscribe_callback function.
• Unsubscribe: Stop listening using unsubscribe_mcp_server.

# Example: Get MCP Server
from v2.nacos.ai.model.ai_param import GetMcpServerParam
mcp_server = await ai_client.get_mcp_server(
    GetMcpServerParam(mcp_name='my-mcp-server', version='1.0.0')
)

# Example: Publish MCP Server
from v2.nacos.ai.model.ai_param import ReleaseMcpServerParam
from v2.nacos.ai.model.mcp.mcp import McpServerBasicInfo, ServerVersionDetail

server_spec = McpServerBasicInfo(
    name='my-mcp-server',
    description='My MCP Server',
    protocol='http',
    versionDetail=ServerVersionDetail(version='1.0.0')
)
result = await ai_client.release_mcp_server(
    ReleaseMcpServerParam(server_spec=server_spec)
)

# Example: Register Endpoint
from v2.nacos.ai.model.ai_param import RegisterMcpServerEndpointParam
await ai_client.register_mcp_server_endpoint(
    RegisterMcpServerEndpointParam(
        mcp_name='my-mcp-server',
        address='127.0.0.1',
        port=8080,
        version='1.0.0'
    )
)

# Example: Subscribe to changes
from v2.nacos.ai.model.ai_param import SubscribeMcpServerParam

async def mcp_listener(mcp_id, namespace_id, mcp_name, mcp_server_detail):
    print(f"MCP Server changed: {mcp_name}, version: {mcp_server_detail.version}")

await ai_client.subscribe_mcp_server(
    SubscribeMcpServerParam(
        mcp_name='my-mcp-server',
        version='1.0.0',
        subscribe_callback=mcp_listener
    )
)

Use ClientConfigBuilder to create a configuration object for Nacos services. This configuration is required to initialize both NacosConfigService and NacosNamingService.

Key Configuration Options

• server_address (required): The Nacos server address (e.g., localhost:8848).
• access_key / secret_key: Aliyun credentials for authentication.
• username / password: Standard Nacos authentication credentials.
• namespace_id: The ID of the namespace to use (defaults to empty string).
• log_level: Logging level (default: logging.INFO).
• grpc_config: Configuration for gRPC communication, including grpc_timeout (default: 3000ms) and port_offset (default: 1000).
• tls_config: Settings for TLS, including enabled, ca_file, cert_file, and key_file.

from v2.nacos import ClientConfigBuilder, GRPCConfig
import os

client_config = (ClientConfigBuilder()
                 .access_key(os.getenv('NACOS_ACCESS_KEY'))
                 .secret_key(os.getenv('NACOS_SECRET_KEY'))
                 .server_address(os.getenv('NACOS_SERVER_ADDR', 'localhost:8848'))
                 .log_level('INFO')
                 .grpc_config(GRPCConfig(grpc_timeout=5000))
                 .build())

The NacosNamingService manages service instances and discovery. Initialize it using create_naming_service.

Key Operations:

• Register Instance: Adds a single instance to a service.
• Batch Register Instances: Registers multiple instances at once using BatchRegisterInstanceParam.
• Deregister Instance: Removes a specific instance.
• Update Instance: Updates metadata or weight of an existing instance.
• Get/List Services: Retrieve service information or list all available services.
• List Instances: List instances belonging to a service (can filter by healthy_only).
• Subscribe Service: Receive callbacks when the instance list for a service changes.
• Unsubscribe Service: Stop receiving callbacks for a service.
• Shutdown: Gracefully stops the client.

from v2.nacos import NacosNamingService, RegisterInstanceParam, SubscribeServiceParam

# Initialize
naming_client = await NacosNamingService.create_naming_service(client_config)

# Register an instance
await naming_client.register_instance(RegisterInstanceParam(
    service_name='my_service',
    group_name='DEFAULT_GROUP',
    ip='1.1.1.1',
    port=7001,
    weight=1.0,
    cluster_name='c1',
    metadata={'a': 'b'},
    enabled=True,
    healthy=True,
    ephemeral=True
))

# Subscribe to service changes
async def on_subscribe(instance_list):
    print(f"Instances changed: {instance_list}")

await naming_client.subscribe(SubscribeServiceParam(
    service_name='my_service',
    group_name='DEFAULT_GROUP',
    subscribe_callback=on_subscribe
))

The NacosNamingService handles service discovery and instance registration. These methods are asynchronous.

Initialize Naming Service

naming_client = await NacosNamingService.create_naming_service(client_config)

Register an Instance

Registers a single service instance with details like IP, port, weight, and metadata.

response = await client.register_instance(
    request=RegisterInstanceParam(
        service_name='nacos.test.1',
        group_name='DEFAULT_GROUP',
        ip='1.1.1.1',
        port=7001,
        weight=1.0,
        cluster_name='c1',
        metadata={'a': 'b'},
        enabled=True,
        healthy=True,
        ephemeral=True
    )
)

Batch Register Instances

Efficiently register multiple instances at once using BatchRegisterInstanceParam.

param1 = RegisterInstanceParam(service_name='svc', group_name='DEFAULT_GROUP', ip='1.1.1.1', port=7001)
param2 = RegisterInstanceParam(service_name='svc', group_name='DEFAULT_GROUP', ip='1.1.1.2', port=7002)

response = await client.batch_register_instances(
    request=BatchRegisterInstanceParam(
        service_name='svc',
        group_name='DEFAULT_GROUP',
        instances=[param1, param2]
    )
)

Deregister/Update Instance

• Deregister: Remove an instance from the registry.
• Update: Modify properties (like weight or metadata) of an existing instance.

# Deregister
await client.deregister_instance(request=DeregisterInstanceParam(service_name='svc', group_name='DEFAULT_GROUP', ip='1.1.1.1', port=7001))

# Update
await client.update_instance(request=RegisterInstanceParam(service_name='svc', group_name='DEFAULT_GROUP', ip='1.1.1.1', port=7001, weight=2.0))

Discover Services and Instances

• get_service: Get details for a specific service.
• list_services: List all available services.
• list_instances: List instances for a service. Use healthy_only=True to filter for healthy instances.

# List healthy instances
instances = await client.list_instances(ListInstanceParam(service_name='nacos.test.1', healthy_only=True))

Subscribe to Service Changes

Subscribe to receive updates when the instance list for a service changes.

async def cb(instance_list: List[Instance]):
    print('Received subscribe callback', str(instance_list))

await client.subscribe(
    SubscribeServiceParam(service_name='nacos.test.1', group_name='DEFAULT_GROUP', subscribe_callback=cb)
)

Shutdown

await client.shutdown()

# Naming service example
naming_client = await NacosNamingService.create_naming_service(client_config)

# Register
await naming_client.register_instance(RegisterInstanceParam(
    service_name='my_service', 
    group_name='DEFAULT_GROUP', 
    ip='127.0.0.1', 
    port=8080
))

# List
instances = await naming_client.list_instances(ListInstanceParam(service_name='my_service'))

The NacosConfigService allows you to retrieve, publish, and listen to configuration changes. Note that these methods are asynchronous and must be awaited.

Initialize Config Service

config_client = await NacosConfigService.create_config_service(client_config)

Get Configuration

Retrieves content for a specific data_id and group. It follows a priority order: local failover dir $\rightarrow$ Nacos server $\rightarrow$ snapshot dir.

content = await config_client.get_config(ConfigParam(
    data_id=data_id,
    group=group
))

Publish/Update Configuration

Creates a new config or updates an existing one. Note: To delete a config, use remove_config instead of setting content to None.

res = await client.publish_config(ConfigParam(
    data_id=dataID,
    group=groupName,
    content="Hello world"
))

Add/Remove Listeners

Register a callback function to be invoked when a configuration item changes or is deleted. If the item already exists on the server, the callback is invoked once immediately upon registration.

async def config_listener(tenant, data_id, group, content):
    print(f"Change detected: {content}")

await config_client.add_listener(dataID, groupName, config_listener)

Remove Configuration

res = await client.remove_config(ConfigParam(
    data_id=dataID,
    group=groupName
))

Shutdown

await client.shutdown()

# Full workflow example
config_client = await NacosConfigService.create_config_service(client_config)

# 1. Publish
await config_client.publish_config(ConfigParam(data_id="test", group="DEFAULT_GROUP", content="val=1))

# 2. Listen
async def my_listener(tenant, data_id, group, content):
    print(content)
await config_client.add_listener("test", "DEFAULT_GROUP", my_listener)

# 3. Get
val = await config_client.get_config(ConfigParam(data_id="test", group="DEFAULT_GROUP))

Nacos provides capabilities for Model Context Protocol (MCP) Server management, including registration, discovery, and subscription.

• Get MCP Server: Retrieve information using GetMcpServerParam (requires mcp_name, optional version).
• Release MCP Server: Publish or release an MCP server using ReleaseMcpServerParam. Requires server_spec (name, description, protocol, version).
• Register MCP Server Endpoint: Register a specific network endpoint for an MCP server using RegisterMcpServerEndpointParam (requires mcp_name, address, port).
• Subscribe MCP Server: Listen for changes to an MCP server using SubscribeMcpServerParam. Requires a subscribe_callback function.
• Unsubscribe MCP Server: Stop listening to changes using SubscribeMcpServerParam.

# Get MCP Server
from v2.nacos.ai.model.ai_param import GetMcpServerParam
mcp_server = await ai_client.get_mcp_server(
    GetMcpServerParam(mcp_name='my-mcp-server', version='1.0.0')
)

# Register MCP Server Endpoint
from v2.nacos.ai.model.ai_param import RegisterMcpServerEndpointParam
await ai_client.register_mcp_server_endpoint(
    RegisterMcpServerEndpointParam(
        mcp_name='my-mcp-server',
        address='127.0.0.1',
        port=8080,
        version='1.0.0'
    )
)

Nacos supports dynamic registration and discovery of AI Agents based on the A2A protocol via Agent Cards.

• Get Agent Card: Retrieve agent information using GetAgentCardParam (requires agent_name, optional version and registration_type ['url' or 'service']).
• Release Agent Card: Publish an agent card using ReleaseAgentCardParam. Requires an AgentCard object. Supports registration_type and set_as_latest.
• Register Agent Endpoint: Register a network endpoint for an agent using RegisterAgentEndpointParam. Requires agent_name, address, port, and version. Supports transport (default 'JSONRPC'), path, and support_tls.
• Deregister Agent Endpoint: Remove an endpoint using DeregisterAgentEndpointParam.
• Subscribe Agent Card: Listen for agent card changes using SubscribeAgentCardParam with a subscribe_callback.
• Unsubscribe Agent Card: Stop listening to changes using SubscribeAgentCardParam.

# Release Agent Card
from v2.nacos.ai.model.ai_param import ReleaseAgentCardParam
from a2a.types import AgentCard

agent_card = AgentCard(
    name='my-agent',
    version='1.0.0',
    protocol_version='1.0'
)

await ai_client.release_agent_card(
    ReleaseAgentCardParam(
        agent_card=agent_card,
        registration_type='service',
        set_as_latest=True
    )
)