sls_translate_natural_language_to_query
Convert natural language descriptions into SLS (Log Service) query statements, enabling users to search logs without requiring SLS query syntax knowledge.
Instructions
将自然语言转换为SLS查询语句。
## 功能概述
该工具可以将自然语言描述转换为有效的SLS查询语句，便于用户使用自然语言表达查询需求。
## 使用场景
- 当用户不熟悉SLS查询语法时
- 当需要快速构建复杂查询时
- 当需要从自然语言描述中提取查询意图时
## 使用限制
- 仅支持生成SLS查询，不支持其他数据库的SQL如MySQL、PostgreSQL等
- 生成的是查询语句，而非查询结果，需要配合sls_execute_query工具使用
- 如果查询涉及ARMS应用，应优先使用arms_generate_trace_query工具
- 需要对应的 log_sotre 已经设定了索引信息，如果生成的结果里面有字段没有索引或者开启统计，可能会导致查询失败，需要友好的提示用户增加相对应的索引信息
## 最佳实践
- 提供清晰简洁的自然语言描述
- 不要在描述中包含项目或日志库名称
- 如有需要，指定查询的时间范围
- 首次生成的查询可能不完全符合要求，可能需要多次尝试
## 查询示例
- "帮我生成下 XXX 的日志查询语句"
- "查找最近一小时内的错误日志"
Args:
ctx: MCP上下文，用于访问SLS客户端
text: 用于生成查询的自然语言文本
project: SLS项目名称
log_store: SLS日志库名称
region_id: 阿里云区域ID
Returns:
生成的SLS查询语句
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| log_store | Yes | sls log store name | |
| project | Yes | sls project name | |
| region_id | Yes | aliyun region id,region id format like 'xx-xxx',like 'cn-hangzhou' | |
| text | Yes | the natural language text to generate sls log store query |
Implementation Reference
- Handler implementation for the sls_text_to_sql tool, which translates natural language (text) to SLS SQL query using the SLS AI service.
@self.server.tool() @retry( stop=stop_after_attempt(Config.get_retry_attempts()), wait=wait_fixed(Config.RETRY_WAIT_SECONDS), retry=retry_if_exception_type(Exception), reraise=True, ) @handle_tea_exception def sls_text_to_sql( ctx: Context, text: str = Field( ..., description="the natural language text to generate sls log store query", ), project: str = Field(..., description="sls project name"), logStore: str = Field(..., description="sls log store name"), regionId: str = Field( default=..., description="aliyun region id,region id format like 'xx-xxx',like 'cn-hangzhou'", ), ) -> Dict[str, Any]: """将自然语言转换为SLS查询语句。当用户有明确的 logstore 查询需求，必须优先使用该工具来生成查询语句 ## 功能概述 该工具可以将自然语言描述转换为有效的SLS查询语句，便于用户使用自然语言表达查询需求。用户有任何 SLS 日志查询需求时，都需要优先使用该工具。 ## 使用场景 - 当用户不熟悉SLS查询语法时 - 当需要快速构建复杂查询时 - 当需要从自然语言描述中提取查询意图时 ## 使用限制 - 仅支持生成SLS查询，不支持其他数据库的SQL如MySQL、PostgreSQL等 - 生成的是查询语句，而非查询结果，需要配合execute_sql工具使用 - 需要对应的 log_store 已经设定了索引信息，如果生成的结果里面有字段没有索引或者开启统计，可能会导致查询失败，需要友好的提示用户增加相对应的索引信息 ## 最佳实践 - 提供清晰简洁的自然语言描述 - 不要在描述中包含项目或日志库名称 - 如有需要，指定查询的时间范围 - 首次生成的查询可能不完全符合要求，可能需要多次尝试 ## 查询示例 - "帮我生成下 XXX 的日志查询语句" - "查找最近一小时内的错误日志" Args: ctx: MCP上下文，用于访问SLS客户端 text: 用于生成查询的自然语言文本 project: SLS项目名称 logStore: SLS日志库名称 regionId: 阿里云区域ID Returns: 生成的SLS查询语句 """ return utils_text_to_sql(ctx, text, project, logStore, regionId) - Core helper function implementing the translation by invoking Alibaba Cloud SLS CallAiTools API with tool_name='text_to_sql'.
def text_to_sql( ctx: Context, text: str, project: str, log_store: str, region_id: str ) -> dict[str, Any]: logger.info( f"开始文本转SQL查询，输入参数: text={text}, project={project}, log_store={log_store}, region_id={region_id}" ) try: sls_client_wrapper = ctx.request_context.lifespan_context["sls_client"] sls_client: Client = sls_client_wrapper.with_region("cn-shanghai") knowledge_config = sls_client_wrapper.get_knowledge_config(project, log_store) logger.info(f"获取知识库配置: {knowledge_config is not None}") request: CallAiToolsRequest = CallAiToolsRequest() request.tool_name = "text_to_sql" request.region_id = region_id params: dict[str, Any] = { "project": project, "logstore": log_store, "sys.query": append_current_time(text), "external_knowledge_uri": knowledge_config["uri"] if knowledge_config else "", "external_knowledge_key": knowledge_config["key"] if knowledge_config else "", } request.params = params logger.info(f"构建SLS AI工具请求，工具名称: {request.tool_name}") runtime: util_models.RuntimeOptions = util_models.RuntimeOptions() runtime.read_timeout = 60000 runtime.connect_timeout = 60000 tool_response: CallAiToolsResponse = sls_client.call_ai_tools_with_options( request=request, headers={}, runtime=runtime ) data = tool_response.body if "------answer------\n" in data: data = data.split("------answer------\n")[1] result = { "data": data, "requestId": tool_response.headers.get("x-log-requestid", ""), } logger.info(f"文本转SQL查询成功，请求ID: {result['requestId']}") return result except Exception as e: logger.error(f"调用SLS AI工具失败，异常详情: {str(e)}", exc_info=True) raise - src/mcp_server_aliyun_observability/server.py:52-52 (registration)Registration of the IaaS toolkit containing the sls_text_to_sql tool.
register_iaas_tools(mcp_server) - Function that instantiates IaaSToolkit, which registers the sls_text_to_sql tool.
def register_iaas_tools(server: FastMCP): """Register IaaS toolkit tools with the FastMCP server Args: server: FastMCP server instance """ IaaSToolkit(server) - Pydantic input schema validation for the tool parameters: text (natural language), project, logStore, regionId.
ctx: Context, text: str = Field( ..., description="the natural language text to generate sls log store query", ), project: str = Field(..., description="sls project name"), logStore: str = Field(..., description="sls log store name"), regionId: str = Field( default=..., description="aliyun region id,region id format like 'xx-xxx',like 'cn-hangzhou'", ), ) -> Dict[str, Any]: