Metadata-Version: 2.4
Name: nacos-sdk-python
Version: 3.2.0
Summary: Python client for Nacos.
Home-page: https://github.com/nacos-group/nacos-sdk-python
Author: nacos
Author-email: 755063194@qq.com
License: Apache License 2.0
Project-URL: Documentation, https://github.com/nacos-group/nacos-sdk-python
Project-URL: Source, https://github.com/nacos-group/nacos-sdk-python
Project-URL: Nacos Open API Guide, https://nacos.io/en-us/docs/open-api.html
Keywords: nacos,nacos-sdk-python
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiofiles>=24.1.0
Requires-Dist: aiohttp>=3.10.11
Requires-Dist: alibabacloud_kms20160120>=2.2.3
Requires-Dist: alibabacloud_tea_openapi>=0.3.12
Requires-Dist: grpcio>=1.66.1
Requires-Dist: protobuf>=3.20.3
Requires-Dist: psutil>=5.9.5
Requires-Dist: pycryptodome>=3.19.1
Requires-Dist: pydantic>=2.10.4
Requires-Dist: a2a-sdk<1.0.0,>=0.3.20
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: keywords
Dynamic: license
Dynamic: license-file
Dynamic: project-url
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary


# nacos-sdk-python v3

English | [简体中文](README_CN.md)

A Python implementation of Nacos OpenAPI.

see: https://nacos.io/zh-cn/docs/open-API.html

[![Pypi Version](https://badge.fury.io/py/nacos-sdk-python.svg)](https://badge.fury.io/py/nacos-sdk-python)
[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/nacos-group/nacos-sdk-python/blob/master/LICENSE)

### Supported Python version：

Python 3.10+

### Supported Nacos version

Supported Nacos version over 3.x

**Note:** AI Client feature requires Nacos server version 3.1.0 or above.

## Installation

```shell
 pip install nacos-sdk-python
```

## Client Configuration

```python
from v2.nacos import NacosNamingService, NacosConfigService, NacosAIService, ClientConfigBuilder, GRPCConfig, \
    Instance, SubscribeServiceParam, RegisterInstanceParam, DeregisterInstanceParam, \
    BatchRegisterInstanceParam, GetServiceParam, ListServiceParam, ListInstanceParam, ConfigParam
from v2.nacos.ai.model.ai_param import GetPromptParam, SubscribePromptParam, DownloadSkillParam
    
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())
```

* *server_address* - **required**  - Nacos server address
* *access_key* - The aliyun accessKey to authenticate.
* *secret_key* - The aliyun secretKey to authenticate.
* *credentials_provider* - The custom access key manager.
* *username* - The username to authenticate.
* *password* - The password to authenticate.
* *log_level* - Log level | default: `logging.INFO`
* *cache_dir* - cache dir path. | default: `~/nacos/cache`
* *log_dir* - log dir path. | default: `~/logs/nacos`
* *namespace_id* - namespace id.  | default: ``
* *grpc_config* - grpc config.
  * *max_receive_message_length* - max receive message length in grpc.  | default: 100 * 1024 * 1024
  * *max_keep_alive_ms* - max keep alive ms in grpc. | default: 60 * 1000
  * *initial_window_size* - initial window size in grpc.  | default: 10 * 1024 * 1024
  * *initial_conn_window_size* - initial connection window size in grpc. | default: 10 * 1024 * 1024
  * *grpc_timeout* - grpc timeout in milliseconds. | default: 3000
  * *port_offset* - gRPC port offset. The gRPC port = HTTP port + offset. | default: 1000
  * *capability_negotiation_timeout* - timeout for capability negotiation in milliseconds. | default: 5000
* *tls_config* - tls config
  * *enabled* - whether enable tls.
  * *ca_file* - ca file path.
  * *cert_file* - cert file path.
  * *key_file* - key file path.
* *kms_config* - aliyun kms config 
  * *enabled* - whether enable aliyun kms.
  * *endpoint* - aliyun kms endpoint.
  * *access_key* - aliyun accessKey.
  * *secret_key* - aliyun secretKey.
  * *password* - aliyun kms password.

## Config Client

```python
config_client = await NacosConfigService.create_config_service(client_config)
```

### config client common parameters

> `param: ConfigParam`

* `param` *data_id* Data id.
* `param` *group* Group, use `DEFAULT_GROUP` if no group specified.
* `param` *content* Config content.
* `param` *tag* Config tag.
* `param` *app_name* Application name.
* `param` *beta_ips* Beta test ip address.
* `param` *cas_md5* MD5 check code.
* `param` *type* Config type.
* `param` *src_user* Source user.
* `param` *encrypted_data_key* Encrypted data key.
* `param` *kms_key_id* Kms encrypted data key id.
* `param` *usage_type* Usage type.

### Get Config

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

* `param` *ConfigParam* config client common parameters. When getting configuration, it is necessary to specify the
  required data_id and group in param.
* `return` Config content if success or an exception will be raised.

Get value of one config item following priority:

* Step 1 - Get from local failover dir.

* Step 2 - Get from one server until value is got or all servers tried.
    * Content will be saved to snapshot dir after got from server.

* Step 3 - Get from snapshot dir.

### Add Listener

```python
async def config_listener(tenant, data_id, group, content):
    print("listen, tenant:{} data_id:{} group:{} content:{}".format(tenant, data_id, group, content))

await config_client.add_listener(dataID, groupName, config_listener)
```

* `param` *ConfigParam* config client common parameters.
* `listener` *listener* Configure listener, defined by the namespace_id、group、data_id、content.
* `return`

Add Listener to a specified config item.

* Once changes or deletion of the item happened, callback functions will be invoked.
* If the item is already exists in server, callback functions will be invoked for once.
* Callback functions are invoked from current process.

### Remove Listener

```python
await client.remove_listener(dataID, groupName, config_listener)
```

* `param` *ConfigParam* config client common parameters.
* `return` True if success or an exception will be raised.

Remove watcher from specified key.

### Publish Config

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

* `param` *ConfigParam* config client common parameters. When publishing configuration, it is necessary to specify the
  required data_id, group and content in param.
* `return` True if success or an exception will be raised.

Publish one congfig data item to Nacos.

* If the data key is not exist, create one first.
* If the data key is exist, update to the content specified.
* Content can not be set to None, if there is need to delete config item, use function **remove** instead.

### Remove Config

```python
res = await client.remove_config(ConfigParam(
            data_id=dataID,
            group=groupName
        ))
```
* `param` *ConfigParam* config client common parameters.When removing configuration, it is necessary to specify the
  required data_id and group in param.
* `return` True if success or an exception will be raised.

Remove one config data item from Nacos.

### Stop Config Client

```python
await client.shutdown()
```

## Naming Client

```python
naming_client = await NacosNamingService.create_naming_service(client_config)
```

### Register Instance

```python
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 Instance

```python
param1 = 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
                                       )
param2 = RegisterInstanceParam(service_name='nacos.test.1',
                               group_name='DEFAULT_GROUP',
                               ip='1.1.1.1',
                               port=7002,
                               weight=1.0,
                               cluster_name='c1',
                               metadata={'a': 'b'},
                               enabled=True,
                               healthy=True,
                               ephemeral=True
                               )
param3 = RegisterInstanceParam(service_name='nacos.test.1',
                               group_name='DEFAULT_GROUP',
                               ip='1.1.1.1',
                               port=7003,
                               weight=1.0,
                               cluster_name='c1',
                               metadata={'a': 'b'},
                               enabled=True,
                               healthy=False,
                               ephemeral=True
                               )
response = await client.batch_register_instances(
    request=BatchRegisterInstanceParam(service_name='nacos.test.1', group_name='DEFAULT_GROUP',
                                       instances=[param1, param2, param3]))
```

### Deregister Instance

```python
response = await client.deregister_instance(
          request=DeregisterInstanceParam(service_name='nacos.test.1', group_name='DEFAULT_GROUP', ip='1.1.1.1',
                                          port=7001, cluster_name='c1', ephemeral=True)
      )
```

### Update Instance

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

### Get Service

```python
service = await client.get_service(
            GetServiceParam(service_name='nacos.test.1', group_name='DEFAULT_GROUP', cluster_name='c1'))
```

### List Service

```python
service_list = await client.list_services(ListServiceParam())
```

### List Instance

```python
instance_list = await client.list_instances(ListInstanceParam(service_name='nacos.test.1', healthy_only=True))
instance_list = await client.list_instances(ListInstanceParam(service_name='nacos.test.1', healthy_only=False))
instance_list = await client.list_instances(ListInstanceParam(service_name='nacos.test.1', healthy_only=None))
```

### Subscribe

```python
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))
```

### Unsubscribe

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

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

### Stop Naming Client

```python
await client.shutdown()
```

## AI Client

**Important:** AI Client feature requires Nacos server version 3.1.0 or above.

```python
from v2.nacos import NacosAIService, ClientConfigBuilder

client_config = (ClientConfigBuilder()
                 .server_address(os.getenv('NACOS_SERVER_ADDR', 'localhost:8848'))
                 .build())
                 
ai_client = await NacosAIService.create_ai_service(client_config)
```

**Transport Modes:**

- **Prompt** supports both gRPC and HTTP transport. By default, gRPC is used. If the gRPC port is unreachable, the AI client will still start normally (gRPC reconnects asynchronously in the background), and Prompt operations can fall back to HTTP.
- **Skill download** always uses HTTP, regardless of gRPC availability.
- **MCP Server / Agent Card** management uses gRPC.

### MCP Server Management

Nacos provides management capabilities for MCP (Model Context Protocol) Server, including registration, discovery, and subscription, supporting dynamic registration and service discovery of MCP servers.

#### Get MCP Server

```python
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')
)
```

* `param` *GetMcpServerParam* Parameter for retrieving MCP server information.
  * `mcp_name` - Name of the MCP server to query (required).
  * `version` - Version of the MCP server to query (optional).
* `return` McpServerDetailInfo if success or an exception will be raised.

#### Release MCP Server

```python
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)
)
```

* `param` *ReleaseMcpServerParam* Parameter for releasing/publishing MCP server.
  * `server_spec` - Basic information specification for the MCP server (required).
  * `tool_spec` - Tool specification defining the tools provided by MCP server (optional).
  * `mcp_endpoint_spec` - Endpoint specification for MCP server network configuration (optional).
* `return` Server ID if success or an exception will be raised.

#### Register MCP Server Endpoint

```python
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'
    )
)
```

* `param` *RegisterMcpServerEndpointParam* Parameter for registering MCP server endpoint.
  * `mcp_name` - Name of the MCP server (required).
  * `address` - IP address or hostname of the MCP server endpoint (required).
  * `port` - Port number of the MCP server endpoint (required).
  * `version` - Version of the MCP server (optional).

#### Subscribe MCP Server

```python
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
    )
)
```

* `param` *SubscribeMcpServerParam* Parameter for subscribing to MCP server changes.
  * `mcp_name` - Name of the MCP server to subscribe to (required).
  * `version` - Version of the MCP server to subscribe to (optional).
  * `subscribe_callback` - Callback function to handle MCP server changes (required).

#### Unsubscribe MCP Server

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

### Agent Card Management

Nacos provides management capabilities for AI Agent, including registration, discovery, and subscription, supporting dynamic registration and service discovery of Agent Card based on A2A protocol.

#### Get Agent Card

```python
from v2.nacos.ai.model.ai_param import GetAgentCardParam

agent_card = await ai_client.get_agent_card(
    GetAgentCardParam(agent_name='my-agent', version='1.0.0')
)
```

* `param` *GetAgentCardParam* Parameter for retrieving agent card information.
  * `agent_name` - Name of the agent card (required).
  * `version` - Target version, if null or empty, get latest version (optional).
  * `registration_type` - Registration type: 'url' or 'service' (optional).
* `return` AgentCardDetailInfo if success or an exception will be raised.

#### Release Agent Card

```python
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
    )
)
```

* `param` *ReleaseAgentCardParam* Parameter for releasing/publishing agent card.
  * `agent_card` - Agent card information (required).
  * `registration_type` - Registration type: 'url' or 'service' (optional, default: 'service').
  * `set_as_latest` - Whether to set as the latest version (optional, default: False).

#### Register Agent Endpoint

```python
from v2.nacos.ai.model.ai_param import RegisterAgentEndpointParam

await ai_client.register_agent_endpoint(
    RegisterAgentEndpointParam(
        agent_name='my-agent',
        address='127.0.0.1',
        port=8080,
        version='1.0.0',
        transport='JSONRPC',
        path='/agent',
        support_tls=False
    )
)
```

* `param` *RegisterAgentEndpointParam* Parameter for registering agent endpoint.
  * `agent_name` - Name of the agent (required).
  * `address` - IP address or hostname of the agent endpoint (required).
  * `port` - Port number of the agent endpoint (required).
  * `version` - Version of the agent (required).
  * `transport` - Transport protocol (optional, default: 'JSONRPC').
  * `path` - URL path for the endpoint (optional).
  * `support_tls` - Whether TLS is supported (optional, default: False).

#### Deregister Agent Endpoint

```python
from v2.nacos.ai.model.ai_param import DeregisterAgentEndpointParam

await ai_client.deregister_agent_endpoint(
    DeregisterAgentEndpointParam(
        agent_name='my-agent',
        address='127.0.0.1',
        port=8080,
        version='1.0.0'
    )
)
```

* `param` *DeregisterAgentEndpointParam* Parameter for deregistering agent endpoint.
  * `agent_name` - Name of the agent (required).
  * `address` - IP address or hostname of the agent endpoint (required).
  * `port` - Port number of the agent endpoint (required).
  * `version` - Version of the agent (required).

#### Subscribe Agent Card

```python
from v2.nacos.ai.model.ai_param import SubscribeAgentCardParam

async def agent_listener(agent_name, agent_card_detail):
    print(f"Agent card changed: {agent_name}, version: {agent_card_detail.version}")

await ai_client.subscribe_agent_card(
    SubscribeAgentCardParam(
        agent_name='my-agent',
        version='1.0.0',
        subscribe_callback=agent_listener
    )
)
```

* `param` *SubscribeAgentCardParam* Parameter for subscribing to agent card changes.
  * `agent_name` - Name of the agent (required).
  * `version` - Version of the agent (optional).
  * `subscribe_callback` - Callback function to handle agent card changes (required).

#### Unsubscribe Agent Card

```python
await ai_client.unsubscribe_agent_card(
    SubscribeAgentCardParam(
        agent_name='my-agent',
        version='1.0.0',
        subscribe_callback=agent_listener
    )
)
```

### Prompt Management

Nacos provides prompt template management capabilities, including retrieval, subscription, and rendering with variable substitution.

#### Get Prompt

```python
from v2.nacos.ai.model.ai_param import GetPromptParam

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

* `param` *GetPromptParam* Parameter for retrieving prompt information.
  * `prompt_key` - Key of the prompt to query (required).
  * `version` - Version of the prompt (optional).
  * `label` - Label of the prompt (optional).
* `return` Prompt if success or an exception will be raised.

#### Render Prompt with Variables

The `Prompt` object supports template rendering with `{{variableName}}` placeholders. Variables defined in the prompt may include default values via `PromptVariable.defaultValue`. When rendering, default values are applied first, then overridden by user-provided values.

```python
# Render the prompt template with variable substitution
result = prompt.render({"name": "Alice", "place": "Nacos"})
print(result)  # e.g. "Hello Alice, welcome to Nacos!"

# Variables with defaultValue will be used automatically if not overridden
# For example, if the prompt has a variable: PromptVariable(name="lang", defaultValue="en")
# Calling render without providing "lang" will use "en" as the value
result = prompt.render({"name": "Alice"})
```

* `param` *variables* - A dict of variable name to value mappings (optional). Overrides default values defined in `PromptVariable.defaultValue`.
* `return` Rendered string with all `{{variableName}}` placeholders replaced.

#### Subscribe Prompt

```python
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
    )
)
```

* `param` *SubscribePromptParam* Parameter for subscribing to prompt changes.
  * `prompt_key` - Key of the prompt to subscribe to (required).
  * `version` - Version of the prompt (optional).
  * `label` - Label of the prompt (optional).
  * `subscribe_callback` - Callback function to handle prompt changes (required).
* `return` Current Prompt if success or an exception will be raised.

#### Unsubscribe Prompt

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

### Skill Download

Nacos supports downloading skill packages as ZIP archives.

#### Download Skill ZIP

```python
from v2.nacos.ai.model.ai_param import DownloadSkillParam

zip_bytes = await ai_client.download_skill_zip(
    DownloadSkillParam(skill_name='my-skill', version='1.0.0')
)

# Save to file
with open('my-skill.zip', 'wb') as f:
    f.write(zip_bytes)
```

* `param` *DownloadSkillParam* Parameter for downloading a skill ZIP.
  * `skill_name` - Name of the skill (required).
  * `version` - Target skill version (optional, defaults to latest).
  * `label` - Target skill label, e.g. "latest", "stable" (optional).
* `return` ZIP file content as bytes if success or an exception will be raised.

### Stop AI Client

```python
await ai_client.shutdown()
```
