Redis MCP Server

Instructions

Implementation Reference

• src/tools/misc.py:202-256 (handler)

  The core handler implementation for the 'search_redis_documents' tool. Decorated with @mcp.tool() for automatic registration. Performs an HTTP GET request to the configured MCP_DOCS_SEARCH_URL with the question as query parameter 'q', parses JSON response or handles errors appropriately.
  @mcp.tool() async def search_redis_documents( question: str, ) -> Union[List[Dict[str, Any]], Dict[str, Any]]: """Search Redis documentation and knowledge base to learn about Redis concepts and use cases. This tool exposes updated and curated documentation, and must be invoked every time the user wants to learn more in areas including: **Common Use Cases:** - Session Management: User session storage and management - Caching: Application-level and database query caching - Rate Limiting: API throttling and request limiting - Leaderboards: Gaming and ranking systems - Semantic Search: AI-powered similarity search - Agentic Workflows: AI agent state and memory management - RAG (Retrieval-Augmented Generation): Vector storage for AI applications - Real-time Analytics: Counters, metrics, and time-series data - Message Queues: Task queues and job processing - Geospatial: Location-based queries and proximity search Args: question: The question about Redis concepts, data structures, features, or use cases Returns: Union[List[Dict[str, Any]], Dict[str, Any]]: A list of documentation results from the API, or a dict with an error message. """ if not MCP_DOCS_SEARCH_URL: return {"error": "MCP_DOCS_SEARCH_URL environment variable is not configured"} if not question.strip(): return {"error": "Question parameter cannot be empty"} try: headers = { "Accept": "application/json", "User-Agent": f"Redis-MCP-Server/{__version__}", } async with aiohttp.ClientSession() as session: async with session.get( url=MCP_DOCS_SEARCH_URL, params={"q": question}, headers=headers ) as response: # Try to parse JSON response try: result = await response.json() return result except aiohttp.ContentTypeError: # If not JSON, return text content text_content = await response.text() return {"error": f"Non-JSON response: {text_content}"} except aiohttp.ClientError as e: return {"error": f"HTTP client error: {str(e)}"} except Exception as e: return {"error": f"Unexpected error calling ConvAI API: {str(e)}"}