scan_keys

Instructions

Implementation Reference

• src/redis_mcp_server/server.py:70-82 (handler)

  MCP tool handler for 'scan_keys'. Decorated with @mcp.tool(), it accepts optional pattern and count parameters, calls db.scan_keys(), and returns JSON-serialized results.

  @mcp.tool()
  def scan_keys(pattern: str = "*", count: Optional[int] = None) -> str:
      """扫描匹配的键
  
      Args:
          pattern: 匹配模式（如 user:* 、session:* ），默认 *
          count: 最多返回键数量，默认100
      """
      try:
          result = db.scan_keys(pattern, count)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)

• src/redis_mcp_server/db.py:98-118 (helper)

  Core database method RedisConnection.scan_keys(). Uses Redis SCAN command to iterate over keys matching a pattern, respects max_key_count config, filters blocked/disallowed keys, and returns a dict with pattern, keys list, count, and truncated flag.

  def scan_keys(
      self, pattern: str = "*", count: Optional[int] = None
  ) -> dict[str, Any]:
      """扫描匹配的键"""
      max_count = count or self.config.max_key_count
      keys = []
      cursor = 0
      while True:
          cursor, batch = self._client.scan(cursor=cursor, match=pattern, count=min(max_count, 100))
          keys.extend(batch)
          if cursor == 0 or len(keys) >= max_count:
              break
      keys = keys[:max_count]
      # 过滤不允许访问的键
      keys = [k for k in keys if self._is_key_allowed(k)]
      return {
          "pattern": pattern,
          "keys": keys,
          "count": len(keys),
          "truncated": len(keys) >= max_count,
      }

• src/redis_mcp_server/server.py:70-82 (registration)

  Registration via @mcp.tool() decorator on line 70 registers 'scan_keys' as a tool in the FastMCP server.

  @mcp.tool()
  def scan_keys(pattern: str = "*", count: Optional[int] = None) -> str:
      """扫描匹配的键
  
      Args:
          pattern: 匹配模式（如 user:* 、session:* ），默认 *
          count: 最多返回键数量，默认100
      """
      try:
          result = db.scan_keys(pattern, count)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)

• src/redis_mcp_server/server.py:71-77 (schema)

  Input schema defined by function signature: pattern (str, default '*') and count (Optional[int]). Docstring describes parameters.

  def scan_keys(pattern: str = "*", count: Optional[int] = None) -> str:
      """扫描匹配的键
  
      Args:
          pattern: 匹配模式（如 user:* 、session:* ），默认 *
          count: 最多返回键数量，默认100
      """

• tests/test_server.py:100-108 (registration)

  Test confirming 'scan_keys' is registered among expected tool names.

  def test_server_has_tools(self):
      tool_names = list(self.server.mcp._tool_manager._tools.keys())
      expected = [
          "ping", "server_info", "db_size", "scan_keys",
          "get_key_type", "get_key_value", "get_hash",
          "get_memory_usage", "set_string", "delete_keys", "set_expire",
      ]
      for t in expected:
          assert t in tool_names, f"缺少工具: {t}"