xadd
Add entries to Redis streams with optional expiration times to manage time-series data or event logs in Redis databases.
Instructions
Add an entry to a Redis stream with an optional expiration time.
Args: key (str): The stream key. fields (dict): The fields and values for the stream entry. expiration (int, optional): Expiration time in seconds.
Returns: str: The ID of the added entry or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| key | Yes | ||
| fields | Yes | ||
| expiration | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/stream.py:9-32 (handler)The xadd tool handler function implementing the Redis XADD operation with optional expiration handling and error management.
@mcp.tool() async def xadd( key: str, fields: Dict[str, Any], expiration: Optional[int] = None ) -> str: """Add an entry to a Redis stream with an optional expiration time. Args: key (str): The stream key. fields (dict): The fields and values for the stream entry. expiration (int, optional): Expiration time in seconds. Returns: str: The ID of the added entry or an error message. """ try: r = RedisConnectionManager.get_connection() entry_id = r.xadd(key, fields) if expiration: r.expire(key, expiration) return f"Successfully added entry {entry_id} to {key}" + ( f" with expiration {expiration} seconds" if expiration else "" ) except RedisError as e: return f"Error adding to stream {key}: {str(e)}" - src/common/server.py:6-17 (registration)MCP server initialization and tool loading mechanism that imports all src/tools modules, registering the xadd tool via its decorator.
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() - src/tools/stream.py:10-11 (schema)Type hints defining the input schema (key: str, fields: dict[str, Any], expiration: Optional[int]) and output (str) for the xadd tool.
async def xadd( key: str, fields: Dict[str, Any], expiration: Optional[int] = None