surreal-mcp

Instructions

>>> 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}

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