xrange
Retrieve specified entries from a Redis stream by providing the stream key and optional count. Ideal for efficient data extraction and management in Redis systems.
Instructions
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.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| count | No | ||
| key | Yes |
Input Schema (JSON Schema)
{
"properties": {
"count": {
"default": 1,
"title": "Count",
"type": "integer"
},
"key": {
"title": "Key",
"type": "string"
}
},
"required": [
"key"
],
"title": "xrangeArguments",
"type": "object"
}
Implementation Reference
- src/tools/stream.py:35-51 (handler)The handler function implementing the 'xrange' MCP tool. It reads up to 'count' entries from the Redis stream specified by 'key' and returns them as a string or an error message. Registered via @mcp.tool() decorator.@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)}"
- src/common/server.py:6-17 (registration)The registration mechanism loads all tool modules from src/tools, including stream.py which contains the xrange tool, via @mcp.tool() decorators on functions.def load_tools(): import src.tools as tools_pkg for _, module_name, _ in pkgutil.iter_modules(tools_pkg.__path__): importlib.import_module(f"src.tools.{module_name}") # Initialize FastMCP server mcp = FastMCP("Redis MCP Server", dependencies=["redis", "dotenv", "numpy", "aiohttp"]) # Load tools load_tools()