Alibaba Cloud Observability MCP Server

Instructions

Implementation Reference

• src/mcp_server_aliyun_observability/toolkits/iaas/toolkit.py:137-199 (handler)

  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)
• src/mcp_server_aliyun_observability/utils.py:347-403 (helper)

  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)
• src/mcp_server_aliyun_observability/toolkits/iaas/toolkit.py:667-673 (registration)

  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)
• src/mcp_server_aliyun_observability/toolkits/iaas/toolkit.py:146-157 (schema)

  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]: