csmar_query

Instructions

Implementation Reference

• src/index.js:489-518 (registration)

  Registration of the 'csmar_query' tool as an MCP tool. It defines the input schema (table_name, columns, condition, start_time, end_time, limit, format) and the handler that calls the Python client's 'query' action.

  // 6. 通用查询
  server.registerTool(
      'csmar_query',
      {
          description: '通用 CSMAR 数据查询',
          inputSchema: {
              table_name: z.string().describe('表名称'),
              columns: z.array(z.string()).optional().describe('要查询的字段列表'),
              condition: z.string().optional().describe('查询条件 (SQL WHERE 子句)'),
              start_time: z.string().optional().describe('开始时间 (YYYY-MM-DD)'),
              end_time: z.string().optional().describe('结束时间 (YYYY-MM-DD)'),
              limit: z.number().optional().describe('返回记录数限制'),
              format: z.enum(['json', 'dataframe']).optional().describe('返回格式'),
          },
      },
      async ({ table_name, columns = [], condition = '', start_time, end_time, limit, format = 'json' }) => {
          try {
              const loginResult = await ensureLogin();
              if (!loginResult.success) {
                  return { content: [{ type: 'text', text: JSON.stringify(loginResult, null, 2) }], isError: true };
              }
              
              const client = await initPythonClient();
              const result = await client.call('query', { table_name, columns, condition, start_time, end_time, limit, format });
              return { content: [{ type: 'text', text: JSON.stringify(result, null, 2) }] };
          } catch (error) {
              return { content: [{ type: 'text', text: `查询错误: ${error.message}` }], isError: true };
          }
      }
  );

• src/index.js:504-518 (handler)

  The handler function for 'csmar_query'. It ensures login, then calls the persistent Python client's 'query' action with the provided parameters (table_name, columns, condition, start_time, end_time, limit, format) and returns the result as JSON.

      async ({ table_name, columns = [], condition = '', start_time, end_time, limit, format = 'json' }) => {
          try {
              const loginResult = await ensureLogin();
              if (!loginResult.success) {
                  return { content: [{ type: 'text', text: JSON.stringify(loginResult, null, 2) }], isError: true };
              }
              
              const client = await initPythonClient();
              const result = await client.call('query', { table_name, columns, condition, start_time, end_time, limit, format });
              return { content: [{ type: 'text', text: JSON.stringify(result, null, 2) }] };
          } catch (error) {
              return { content: [{ type: 'text', text: `查询错误: ${error.message}` }], isError: true };
          }
      }
  );

• src/index.js:493-503 (schema)

  Input schema for 'csmar_query' using Zod validation: table_name (string, required), columns (array of strings, optional), condition (string, optional), start_time (string, optional), end_time (string, optional), limit (number, optional), format (enum 'json' or 'dataframe', optional).

      description: '通用 CSMAR 数据查询',
      inputSchema: {
          table_name: z.string().describe('表名称'),
          columns: z.array(z.string()).optional().describe('要查询的字段列表'),
          condition: z.string().optional().describe('查询条件 (SQL WHERE 子句)'),
          start_time: z.string().optional().describe('开始时间 (YYYY-MM-DD)'),
          end_time: z.string().optional().describe('结束时间 (YYYY-MM-DD)'),
          limit: z.number().optional().describe('返回记录数限制'),
          format: z.enum(['json', 'dataframe']).optional().describe('返回格式'),
      },
  },

• src/python_client.py:275-296 (helper)

  The Python-side 'query' method on CSMARClient that actually executes the CSMAR SDK query. It calls csmar.query() or csmar.query_df() based on format, applies limit, and returns the data.

  def query(self, columns: list, condition: str, table_name: str,
           start_time: Optional[str] = None, end_time: Optional[str] = None,
           format: str = "json", limit: Optional[int] = None) -> Dict[str, Any]:
      try:
          csmar = self._ensure_csmar()
          if format == "dataframe":
              data = csmar.query_df(columns, condition, table_name, start_time, end_time)
              result = data.to_dict('records') if hasattr(data, 'to_dict') else data
          else:
              data = csmar.query(columns, condition, table_name, start_time, end_time)
              result = data
  
          if result is None:
              return {"success": True, "table": table_name, "data": [], "count": 0, "message": "查询结果为空"}
  
          if limit and isinstance(result, list):
              result = result[:limit]
  
          return {"success": True, "table": table_name, "data": result,
                  "count": len(result) if isinstance(result, list) else 1}
      except Exception as e:
          return {"success": False, "error": f"查询数据失败: {str(e)}"}

• src/python_client.py:324-327 (helper)

  The command dispatch in python_client.py that maps the 'query' action string to the CSMARClient.query() method, passing through all parameters from the Node.js side.

  "query": lambda: client.query(
      params.get("columns", []), params.get("condition", ""), params.get("table_name", ""),
      params.get("start_time"), params.get("end_time"), params.get("format", "json"), params.get("limit")
  ),