Redis MCP Server

Instructions

Implementation Reference

• src/tools/stream.py:35-51 (handler)

  The 'xrange' tool handler: an async function decorated with @mcp.tool() that reads up to 'count' entries from the Redis stream 'key' and returns them as a string or an error message.

  @mcp.tool()
  async def xrange(key: str, count: int = 1) -> str:
      """Read entries from a Redis stream.
  
      Args:
          key (str): The stream key.
          count (int, optional): Number of entries to retrieve.
  
      Returns:
          str: The retrieved stream entries or an error message.
      """
      try:
          r = RedisConnectionManager.get_connection()
          entries = r.xrange(key, count=count)
          return str(entries) if entries else f"Stream {key} is empty or does not exist"
      except RedisError as e:
          return f"Error reading from stream {key}: {str(e)}"