get_key_value

Instructions

Implementation Reference

• src/redis_mcp_server/db.py:234-250 (handler)

  Core handler: RedisConnection.get_key_value() - auto-detects key type and delegates to the appropriate type-specific getter (string/list/set/hash/zset). This is the actual business logic implementation.

  def get_key_value(self, key: str) -> dict[str, Any]:
      """自动识别键类型并获取值"""
      key_info = self.get_key_type(key)
      key_type = key_info["type"]
  
      if key_type == "string":
          return self.get_string(key)
      elif key_type == "list":
          return self.get_list(key)
      elif key_type == "set":
          return self.get_set(key)
      elif key_type == "hash":
          return self.get_hash(key)
      elif key_type == "zset":
          return self.get_zset(key)
      else:
          return {"key": key, "type": key_type, "note": f"暂不支持 {key_type} 类型的自动读取"}

• src/redis_mcp_server/server.py:101-114 (registration)

  MCP tool registration via @mcp.tool() decorator wrapping get_key_value, with docstring describing it auto-detects key type and fetches value.

  @mcp.tool()
  def get_key_value(key: str) -> str:
      """自动识别键类型并获取值（支持string/list/set/hash/zset）
  
      Args:
          key: Redis键名
      """
      try:
          result = db.get_key_value(key)
          return json.dumps(result, ensure_ascii=False, indent=2, default=str)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)

• src/redis_mcp_server/server.py:101-232 (schema)

  Input schema: accepts 'key: str' parameter. Output schema: returns JSON string with key, type, value (or fields/members for complex types), and metadata.

  @mcp.tool()
  def get_key_value(key: str) -> str:
      """自动识别键类型并获取值（支持string/list/set/hash/zset）
  
      Args:
          key: Redis键名
      """
      try:
          result = db.get_key_value(key)
          return json.dumps(result, ensure_ascii=False, indent=2, default=str)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  @mcp.tool()
  def get_hash(key: str) -> str:
      """获取哈希类型的所有字段和值
  
      Args:
          key: Redis键名
      """
      try:
          result = db.get_hash(key)
          return json.dumps(result, ensure_ascii=False, indent=2, default=str)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  @mcp.tool()
  def get_memory_usage(key: str) -> str:
      """获取键的内存使用量
  
      Args:
          key: Redis键名
      """
      try:
          result = db.get_memory_usage(key)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  @mcp.tool()
  def set_string(key: str, value: str, ex: Optional[int] = None) -> str:
      """设置字符串值（需要关闭只读模式）
  
      Args:
          key: Redis键名
          value: 要设置的值
          ex: 过期时间（秒），不填则永不过期
      """
      try:
          result = db.set_string(key, value, ex)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  @mcp.tool()
  def delete_keys(keys: list[str]) -> str:
      """删除键（需要关闭只读模式）
  
      Args:
          keys: 要删除的键名列表
      """
      try:
          result = db.delete_keys(keys)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  @mcp.tool()
  def set_expire(key: str, seconds: int) -> str:
      """设置键的过期时间（需要关闭只读模式）
  
      Args:
          key: Redis键名
          seconds: 过期时间（秒）
      """
      try:
          result = db.set_expire(key, seconds)
          return json.dumps(result, ensure_ascii=False, indent=2)
      except PermissionError as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)
  
  
  # ============================================================
  # 提示词（Prompts）
  # ============================================================
  
  
  @mcp.prompt()
  def analyze_redis() -> str:
      """生成Redis分析提示词"""
      return """请分析Redis实例的状态：
  
  1. 使用 server_info 获取服务器信息
  2. 使用 db_size 查看键数量
  3. 使用 scan_keys 查看键的分布模式
  4. 对重要键使用 get_key_type 和 get_key_value 查看内容
  5. 使用 get_memory_usage 分析内存占用
  6. 总结Redis的使用情况、内存分布、键模式
  7. 提出优化建议（如有）
  
  请用中文回答。"""
  
  
  @mcp.prompt()
  def find_keys_by_pattern(pattern: str) -> str:
      """生成键搜索提示词"""
      return f"""请搜索Redis中匹配 '{pattern}' 的键：
  
  1. 使用 scan_keys 搜索匹配的键
  2. 对每个键使用 get_key_type 查看类型
  3. 对重要键使用 get_key_value 查看值
  4. 使用 get_memory_usage 查看内存占用
  5. 总结搜索结果
  
  请用中文回答。"""

• src/redis_mcp_server/db.py:120-130 (helper)

  Helper: get_key_type() - called by get_key_value to determine the Redis type before dispatching.

  def get_key_type(self, key: str) -> dict[str, Any]:
      """获取键的类型"""
      if not self._is_key_allowed(key):
          raise PermissionError(f"键 '{key}' 不允许访问")
      key_type = self._client.type(key)
      ttl = self._client.ttl(key)
      return {
          "key": key,
          "type": key_type,
          "ttl": ttl if ttl >= 0 else ("永不过期" if ttl == -1 else "键不存在"),
      }