xrange
Retrieve entries from a Redis stream. Specify the stream key and an optional count to limit results.
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
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| key | Yes | ||
| count | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/stream.py:35-51 (handler)The xrange tool handler function that reads entries from a Redis stream using redis-py's xrange() and returns them as a string.
@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/tools/stream.py:35-36 (registration)The @mcp.tool() decorator registers xrange as an MCP tool. The MCP server is initialized in src/common/server.py and auto-loads all tools from src/tools/.
@mcp.tool() async def xrange(key: str, count: int = 1) -> str: - src/tools/stream.py:36-51 (schema)Type annotations define the input schema: key (str) and count (int, default=1). The return type is str.
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:1-6 (helper)The FastMCP server instance 'mcp' that hosts the @mcp.tool() decorator, and the load_tools() function that auto-discovers all tool modules.
import importlib import pkgutil from mcp.server.fastmcp import FastMCP def load_tools(): - src/common/server.py:6-19 (helper)load_tools() dynamically imports all modules in src/tools/, which causes the @mcp.tool() decorators to execute and register the tools.
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", "python-dotenv", "numpy", "aiohttp"] ) # Load tools load_tools()