surreal-mcp

Instructions

Implementation Reference

• surreal_mcp/server.py:190-260 (handler)

  The core handler function for the 'select' MCP tool. It builds a SurrealQL SELECT query from the table and optional id parameters, resolves the namespace and database context, executes the query using repo_query, formats the result as a list with success and count, and handles errors.
  @mcp.tool() async def select( table: str, id: Optional[str] = None, namespace: Optional[str] = None, database: Optional[str] = None, ) -> Dict[str, Any]: """ Select all records from a table or a specific record by ID. This tool provides a simple way to retrieve data from SurrealDB tables. Use this when you need to: - Fetch all records from a table - Retrieve a specific record by its ID - Get data for display or further processing Args: table: The name of the table to select from (e.g., "user", "product", "order") id: Optional ID of a specific record to select. Can be: - Just the ID part (e.g., "john") - will be combined with table name - Full record ID (e.g., "user:john") - will be used as-is - None/omitted - selects all records from the table namespace: Optional SurrealDB namespace override. If not provided, uses SURREAL_NAMESPACE env var. database: Optional SurrealDB database override. If not provided, uses SURREAL_DATABASE env var. Returns: A dictionary containing: - success: Boolean indicating if the selection was successful - data: Array of records (even for single record selection) - count: Number of records returned - error: Error message if selection failed (only present on failure) Examples: >>> await select("user") # Get all users {"success": true, "data": [...], "count": 42} >>> await select("user", "john") # Get specific user {"success": true, "data": [{"id": "user:john", "name": "John Doe", ...}], "count": 1} >>> await select("product", "product:laptop-123") # Using full ID {"success": true, "data": [{"id": "product:laptop-123", ...}], "count": 1} """ try: ns, db = resolve_namespace_database(namespace, database) # Build the query based on whether ID is provided if id: # Handle both "id" and "table:id" formats if ":" in id and id.startswith(f"{table}:"): record_id = id else: record_id = f"{table}:{id}" query_str = f"SELECT * FROM {record_id}" else: query_str = f"SELECT * FROM {table}" logger.info(f"Executing select: {query_str}") result = await repo_query(query_str, namespace=ns, database=db) # Ensure result is always a list if not isinstance(result, list): result = [result] if result else [] return { "success": True, "data": result, "count": len(result) } except Exception as e: logger.error(f"Select failed for {table}: {str(e)}") raise Exception(f"Failed to select from {table}: {str(e)}")
• surreal_mcp/database/__init__.py:64-91 (helper)

  Helper function repo_query that executes the actual SurrealQL SELECT query generated by the select tool handler, using a pooled or override database connection, and parses RecordIDs to strings.
  async def repo_query( query_str: str, vars: Optional[Dict[str, Any]] = None, namespace: Optional[str] = None, database: Optional[str] = None, ) -> List[Dict[str, Any]]: """Execute a SurrealQL query and return the results. Args: query_str: The SurrealQL query to execute vars: Optional variables for the query namespace: Optional namespace override (uses env var if not provided) database: Optional database override (uses env var if not provided) Returns: The query results as a list of dictionaries """ async with db_connection(namespace, database) as connection: try: result = parse_record_ids(await connection.query(query_str, vars)) if isinstance(result, str): raise RuntimeError(result) return result except Exception as e: logger.error(f"Query: {query_str[:200]} vars: {vars}") logger.exception(e) raise
• surreal_mcp/server.py:55-99 (helper)

  Helper function used by the select tool (and others) to resolve namespace and database from tool parameters or fallback to environment variables, returning None,None for pooled connection or specific values for overrides.
  def resolve_namespace_database( namespace: Optional[str] = None, database: Optional[str] = None, ) -> Tuple[Optional[str], Optional[str]]: """ Resolve namespace and database values from parameters or environment variables. Args: namespace: Optional namespace parameter from tool call database: Optional database parameter from tool call Returns: Tuple of (resolved_namespace, resolved_database). Both will be None if using default pooled connection, or both will be strings if using override connection. Raises: ValueError: If namespace/database cannot be determined from either source """ # Get values from env vars as fallback env_namespace = os.environ.get("SURREAL_NAMESPACE") env_database = os.environ.get("SURREAL_DATABASE") # Resolve final values final_namespace = namespace if namespace is not None else env_namespace final_database = database if database is not None else env_database # If both are from env vars (or both params are None), use pooled connection if namespace is None and database is None and env_namespace and env_database: return None, None # Signal to use pooled connection # If either param is provided, we need both values resolved if final_namespace is None or final_database is None: missing = [] if final_namespace is None: missing.append("namespace") if final_database is None: missing.append("database") raise ValueError( f"Missing required database configuration: {', '.join(missing)}. " "Either set SURREAL_NAMESPACE/SURREAL_DATABASE environment variables " "or provide namespace/database parameters in the tool call." ) return final_namespace, final_database