feat(ai): add Prompt management and HTTP transport support (#307)

- Add Prompt model with template variable rendering ({{var}} syntax)
- Add gRPC and HTTP transport for Prompt queries (QueryPromptRequest/Response)
- Add AiHttpClientProxy for HTTP-based AI operations (GET /v3/client/ai/prompt)
- Add PromptCacheHolder with MD5 conditional polling and change notification
- Add PromptSubscribeManager for prompt change subscriptions
- Support transport mode selection via ClientConfig.ai_transport_mode (grpc/http)
- Register QueryPromptResponse in GrpcUtils.remote_type
- Bump version to 3.2.0b1

Author: Sunrisea

setup.py

@@ -53,7 +53,7 @@ def run(self):
 
 setup(
     name="nacos-sdk-python",
-    version="3.0.4",
+    version="3.2.0b1",
     packages=find_packages(
         exclude=["test", "*.tests", "*.tests.*", "tests.*", "tests"]),
     url="https://github.com/nacos-group/nacos-sdk-python",

v2/nacos/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "3.0.4"
+__version__ = "3.2.0b1"
 
 from .common.client_config import (KMSConfig,
                                    GRPCConfig,

v2/nacos/ai/model/ai_constant.py

@@ -23,4 +23,11 @@ class AIConstants:
 
 	A2A_ENDPOINT_TYPE_SERVICE = "SERVICE"
 
-	A2A_ENDPOINT_DEFAULT_TRANSPORT = "JSONRPC"
+	A2A_ENDPOINT_DEFAULT_TRANSPORT = "JSONRPC"
+
+	# AI transport mode: "grpc" (default) or "http"
+	AI_TRANSPORT_MODE_GRPC = "grpc"
+	AI_TRANSPORT_MODE_HTTP = "http"
+
+	# Default prompt cache update interval in seconds
+	DEFAULT_PROMPT_CACHE_UPDATE_INTERVAL = 10

v2/nacos/ai/model/ai_param.py

@@ -7,6 +7,7 @@
 from v2.nacos.ai.model.ai_constant import AIConstants
 from v2.nacos.ai.model.mcp.mcp import McpServerBasicInfo, McpToolSpecification, \
 	McpEndpointSpec, McpServerDetailInfo
+from v2.nacos.ai.model.prompt.prompt import Prompt
 
 
 class GetMcpServerParam(BaseModel):
@@ -97,4 +98,22 @@ class SubscribeAgentCardParam(BaseModel):
 	version: Optional[str] = None
 	# Parameters: agent_name, agent_card_detail_info
 	subscribe_callback: Optional[
-		Callable[[str, AgentCardDetailInfo], Awaitable[None]]] = None
+		Callable[[str, AgentCardDetailInfo], Awaitable[None]]] = None
+
+
+# ==================== Prompt Params ====================
+
+class GetPromptParam(BaseModel):
+	"""Parameter model for retrieving prompt information"""
+	prompt_key: str
+	version: Optional[str] = None
+	label: Optional[str] = None
+
+
+class SubscribePromptParam(BaseModel):
+	"""Parameter model for subscribing to prompt changes"""
+	prompt_key: str
+	version: Optional[str] = None
+	label: Optional[str] = None
+	subscribe_callback: Optional[
+		Callable[[str, Prompt], Awaitable[None]]] = None

v2/nacos/ai/model/ai_request.py

@@ -131,3 +131,27 @@ class ReleaseMcpServerRequest(AbstractMcpRequest):
 	def get_request_type(self) -> str:
 		"""Returns the release MCP server request type"""
 		return RELEASE_MCP_SERVER_REQUEST_TYPE
+
+
+# ==================== Prompt Requests ====================
+
+QUERY_PROMPT_REQUEST_TYPE = "QueryPromptRequest"
+
+
+class AbstractPromptRequest(AbstractAIRequest, ABC):
+	"""Abstract base class for Prompt requests"""
+	namespaceId: Optional[str] = None
+	promptKey: Optional[str] = None
+
+	def get_request_type(self) -> str:
+		raise NotImplementedError("Subclasses should implement this method.")
+
+
+class QueryPromptRequest(AbstractPromptRequest):
+	"""Request for querying prompt by key, version, or label"""
+	version: Optional[str] = None
+	label: Optional[str] = None
+	md5: Optional[str] = None
+
+	def get_request_type(self) -> str:
+		return QUERY_PROMPT_REQUEST_TYPE

v2/nacos/ai/model/ai_response.py

@@ -2,6 +2,7 @@
 
 from v2.nacos.ai.model.a2a.a2a import AgentCardDetailInfo
 from v2.nacos.ai.model.mcp.mcp import McpServerDetailInfo
+from v2.nacos.ai.model.prompt.prompt import Prompt
 from v2.nacos.transport.model.rpc_response import Response
 
 
@@ -58,3 +59,11 @@ def get_response_type(self) -> str:
 		"""Returns the release agent card response type"""
 		return "ReleaseAgentCardResponse"
 
+
+class QueryPromptResponse(Response):
+	"""Response for prompt query requests"""
+	promptInfo: Optional[Prompt] = None
+
+	def get_response_type(self) -> str:
+		return "QueryPromptResponse"
+

v2/nacos/ai/model/cache/prompt_cache_holder.py

@@ -0,0 +1,134 @@
+import asyncio
+import logging
+from typing import Dict, Optional
+
+from v2.nacos.ai.model.cache.prompt_subscribe_manager import PromptSubscribeManager
+from v2.nacos.ai.model.prompt.prompt import Prompt
+from v2.nacos.ai.util.prompt_util import build_prompt_cache_key
+from v2.nacos.common.constants import Constants
+from v2.nacos.common.nacos_exception import NacosException, NOT_FOUND, NOT_MODIFIED
+
+DEFAULT_PROMPT_UPDATE_INTERVAL = 10
+
+
+class PromptCacheHolder:
+
+	def __init__(self, subscribe_manager: PromptSubscribeManager, ai_client_proxy,
+			update_interval: float = DEFAULT_PROMPT_UPDATE_INTERVAL):
+		"""
+		Args:
+			subscribe_manager: Manager for prompt change subscribers
+			ai_client_proxy: Any object with async query_prompt(prompt_key, version, label, md5) method.
+				Can be AIGRPCClientProxy or AiHttpClientProxy.
+			update_interval: Poll interval in seconds (default 10s)
+		"""
+		self.prompt_subscribe_manager = subscribe_manager
+		self.ai_client_proxy = ai_client_proxy
+		self.update_interval = update_interval
+		self.prompt_cache: Dict[str, Prompt] = {}
+		self.cache_lock = asyncio.Lock()
+		self.logger = logging.getLogger(Constants.AI_MODULE)
+		self.tasks: Dict[str, asyncio.Task] = {}
+
+	async def shutdown(self):
+		for task in self.tasks.values():
+			task.cancel()
+		for task in self.tasks.values():
+			try:
+				await task
+			except asyncio.CancelledError:
+				pass
+		self.tasks.clear()
+
+	async def subscribe_prompt(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]) -> Optional[Prompt]:
+		cache_key = build_prompt_cache_key(prompt_key, version, label)
+		prompt = None
+		try:
+			prompt = await self.ai_client_proxy.query_prompt(prompt_key, version, label, None)
+			self._process_prompt(prompt_key, cache_key, prompt)
+		except NacosException as e:
+			if e.error_code != NOT_FOUND:
+				raise
+			self._process_prompt(prompt_key, cache_key, None)
+
+		self._add_update_task(prompt_key, version, label)
+		self.logger.info(f"Subscribed prompt: {prompt_key}, version: {version}, label: {label}")
+		return prompt
+
+	def unsubscribe_prompt(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]):
+		cache_key = build_prompt_cache_key(prompt_key, version, label)
+		self._remove_update_task(prompt_key, version, label)
+		self.prompt_cache.pop(cache_key, None)
+		self.logger.info(f"Unsubscribed prompt: {prompt_key}, version: {version}, label: {label}")
+
+	def get_cached_prompt(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]) -> Optional[Prompt]:
+		cache_key = build_prompt_cache_key(prompt_key, version, label)
+		return self.prompt_cache.get(cache_key)
+
+	def _add_update_task(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]):
+		cache_key = build_prompt_cache_key(prompt_key, version, label)
+		if cache_key not in self.tasks:
+			self.tasks[cache_key] = asyncio.create_task(
+				self._update_prompt_loop(prompt_key, version, label, cache_key))
+
+	def _remove_update_task(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]):
+		cache_key = build_prompt_cache_key(prompt_key, version, label)
+		task = self.tasks.pop(cache_key, None)
+		if task is not None:
+			task.cancel()
+
+	async def _update_prompt_loop(self, prompt_key: str, version: Optional[str],
+			label: Optional[str], cache_key: str):
+		while True:
+			try:
+				await asyncio.sleep(self.update_interval)
+			except asyncio.CancelledError:
+				return
+
+			try:
+				current_prompt = self.prompt_cache.get(cache_key)
+				current_md5 = current_prompt.md5 if current_prompt else None
+				latest_prompt = await self.ai_client_proxy.query_prompt(
+					prompt_key, version, label, current_md5)
+				self._process_prompt(prompt_key, cache_key, latest_prompt)
+			except NacosException as e:
+				if e.error_code == NOT_FOUND:
+					self._process_prompt(prompt_key, cache_key, None)
+				elif e.error_code == NOT_MODIFIED:
+					pass
+				else:
+					self.logger.warning(
+						f"Prompt updater query failed: promptKey={prompt_key}, err={e.message}")
+			except asyncio.CancelledError:
+				return
+			except Exception as e:
+				self.logger.warning(
+					f"Prompt updater unexpected error: promptKey={prompt_key}, err={e}")
+
+	def _process_prompt(self, prompt_key: str, cache_key: str,
+			new_prompt: Optional[Prompt]):
+		old_prompt = self.prompt_cache.get(cache_key)
+		if new_prompt is None:
+			self.prompt_cache.pop(cache_key, None)
+		else:
+			self.prompt_cache[cache_key] = new_prompt
+
+		if self._is_prompt_changed(old_prompt, new_prompt):
+			subscribers = self.prompt_subscribe_manager.subscribers.get(cache_key, [])
+			for callback_func in subscribers:
+				try:
+					asyncio.ensure_future(callback_func(prompt_key, new_prompt))
+				except Exception as e:
+					self.logger.error(f"Prompt change callback error: {e}")
+
+	@staticmethod
+	def _is_prompt_changed(old_prompt: Optional[Prompt],
+			new_prompt: Optional[Prompt]) -> bool:
+		old_json = "" if old_prompt is None else old_prompt.model_dump_json()
+		new_json = "" if new_prompt is None else new_prompt.model_dump_json()
+		return old_json != new_json

v2/nacos/ai/model/cache/prompt_subscribe_manager.py

@@ -0,0 +1,41 @@
+import asyncio
+from typing import Callable, Awaitable, List, Optional
+
+from v2.nacos.ai.model.prompt.prompt import Prompt
+from v2.nacos.ai.util.prompt_util import build_prompt_cache_key
+
+
+class PromptSubscribeManager:
+
+	def __init__(self):
+		self.subscribers: dict[str, List[Callable[[str, Prompt], Awaitable[None]]]] = {}
+		self.lock = asyncio.Lock()
+
+	async def register_subscriber(self, prompt_key: str, version: Optional[str],
+			label: Optional[str],
+			callback_func: Callable[[str, Prompt], Awaitable[None]]):
+		key = build_prompt_cache_key(prompt_key, version, label)
+		async with self.lock:
+			if key not in self.subscribers:
+				self.subscribers[key] = []
+			self.subscribers[key].append(callback_func)
+
+	async def deregister_subscriber(self, prompt_key: str, version: Optional[str],
+			label: Optional[str],
+			callback_func: Callable[[str, Prompt], Awaitable[None]]):
+		if not callback_func:
+			return
+		key = build_prompt_cache_key(prompt_key, version, label)
+		async with self.lock:
+			if key not in self.subscribers:
+				return
+			self.subscribers[key] = [func for func in self.subscribers[key] if func != callback_func]
+			if not self.subscribers[key]:
+				del self.subscribers[key]
+
+	def is_subscribed(self, prompt_key: str, version: Optional[str],
+			label: Optional[str]) -> bool:
+		key = build_prompt_cache_key(prompt_key, version, label)
+		if key not in self.subscribers:
+			return False
+		return len(self.subscribers[key]) > 0

v2/nacos/ai/model/prompt/__init__.py

@@ -0,0 +1,38 @@
+from typing import Optional, Dict
+
+from pydantic import BaseModel
+
+
+class Prompt(BaseModel):
+	"""Prompt entity for AI Prompt management.
+
+	Prompt is stored as a Nacos configuration with fixed group "nacos-ai-prompt"
+	and dataId "{promptKey}.json". The content is stored as JSON format.
+	"""
+
+	promptKey: Optional[str] = None
+	version: Optional[str] = None
+	template: Optional[str] = None
+	md5: Optional[str] = None
+
+	def render(self, variables: Optional[Dict[str, str]] = None) -> Optional[str]:
+		"""Render the prompt template by replacing {{variableName}} with values.
+
+		Example:
+			prompt = Prompt(template="Hello {{name}}, welcome to {{place}}!")
+			result = prompt.render({"name": "Alice", "place": "Nacos"})
+			# Result: "Hello Alice, welcome to Nacos!"
+		"""
+		if self.template is None:
+			return None
+		if not variables:
+			return self.template
+
+		result = self.template
+		for key, value in variables.items():
+			placeholder = "{{" + key + "}}"
+			result = result.replace(placeholder, value if value is not None else "")
+		return result
+
+	def __str__(self):
+		return f"Prompt(promptKey='{self.promptKey}', version='{self.version}')"