unsubscribe
Unsubscribe from a Redis channel by providing its name to stop receiving messages from that channel.
Instructions
Unsubscribe from a Redis channel.
Args: channel: The Redis channel to unsubscribe from.
Returns: A success message or an error message.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| channel | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/tools/pub_sub.py:45-61 (handler)The 'unsubscribe' tool handler function. Decorated with @mcp.tool(), it gets a Redis connection, creates a pubsub object, and unsubscribes from the given channel.
@mcp.tool() async def unsubscribe(channel: str) -> str: """Unsubscribe from a Redis channel. Args: channel: The Redis channel to unsubscribe from. Returns: A success message or an error message. """ try: r = RedisConnectionManager.get_connection() pubsub = r.pubsub() pubsub.unsubscribe(channel) return f"Unsubscribed from channel '{channel}'." except RedisError as e: return f"Error unsubscribing from channel '{channel}': {str(e)}" - src/tools/pub_sub.py:45-61 (schema)Input schema for unsubscribe: expects a single 'channel' (string) parameter. Returns a string.
@mcp.tool() async def unsubscribe(channel: str) -> str: """Unsubscribe from a Redis channel. Args: channel: The Redis channel to unsubscribe from. Returns: A success message or an error message. """ try: r = RedisConnectionManager.get_connection() pubsub = r.pubsub() pubsub.unsubscribe(channel) return f"Unsubscribed from channel '{channel}'." except RedisError as e: return f"Error unsubscribing from channel '{channel}': {str(e)}" - src/tools/pub_sub.py:45-46 (registration)Registered via the @mcp.tool() decorator on line 45. The mcp instance (FastMCP) is defined in src/common/server.py, and modules are auto-loaded by load_tools().
@mcp.tool() async def unsubscribe(channel: str) -> str: - src/common/server.py:6-19 (registration)load_tools() dynamically imports all modules under src.tools, including pub_sub.py, which triggers the @mcp.tool() decorator registration.
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()