xack
Acknowledge processed entries in a Redis stream consumer group by providing the stream key, group name, and entry IDs. Returns confirmation or error.
Instructions
Acknowledge entries that were processed by a consumer group.
Args: key (str): The stream key. group_name (str): The consumer group name. entry_ids (List[str]): Entry IDs to acknowledge.
Returns: str: Confirmation message or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| key | Yes | ||
| group_name | Yes | ||
| entry_ids | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/stream.py:185-211 (handler)The xack tool handler function. Decorated with @mcp.tool(), it acknowledges processed entries in a Redis stream consumer group. Takes key, group_name, and entry_ids parameters, validates entry_ids is non-empty, calls Redis XACK, and returns a confirmation or error message.
@mcp.tool() async def xack(key: str, group_name: str, entry_ids: List[str]) -> str: """Acknowledge entries that were processed by a consumer group. Args: key (str): The stream key. group_name (str): The consumer group name. entry_ids (List[str]): Entry IDs to acknowledge. Returns: str: Confirmation message or an error message. """ if not entry_ids: return "At least one entry ID is required to acknowledge stream entries" try: r = RedisConnectionManager.get_connection() acknowledged = r.xack(key, group_name, *entry_ids) return ( f"Successfully acknowledged {acknowledged} entr" f"{'y' if acknowledged == 1 else 'ies'} in group '{group_name}' on stream '{key}'" ) except RedisError as e: return ( f"Error acknowledging entries for consumer group '{group_name}' on stream " f"'{key}': {str(e)}" ) - src/tools/stream.py:185-186 (registration)Registration of xack as an MCP tool via the @mcp.tool() decorator in src/tools/stream.py. The module is auto-discovered and loaded by load_tools() in src/common/server.py.
@mcp.tool() async def xack(key: str, group_name: str, entry_ids: List[str]) -> str: - src/tools/stream.py:186-196 (schema)Type annotations define the input schema: key (str), group_name (str), entry_ids (List[str]). The docstring serves as the description for the tool.
async def xack(key: str, group_name: str, entry_ids: List[str]) -> str: """Acknowledge entries that were processed by a consumer group. Args: key (str): The stream key. group_name (str): The consumer group name. entry_ids (List[str]): Entry IDs to acknowledge. Returns: str: Confirmation message or an error message. """