surreal-mcp

Instructions

Implementation Reference

• surreal_mcp/server.py:104-151 (handler)

  Primary handler for the 'query' MCP tool. Decorated with @mcp.tool() for registration. Resolves namespace/database, logs the query, calls repo_query to execute, and formats the response with success/data.
  @mcp.tool() async def query( query_string: str, namespace: Optional[str] = None, database: Optional[str] = None, ) -> Dict[str, Any]: """ Execute a raw SurrealQL query against the connected SurrealDB database. This tool allows you to run any valid SurrealQL query directly. Use this for complex queries that don't fit the other tool patterns, such as: - Complex SELECT queries with JOINs, GROUP BY, or aggregations - Custom DEFINE statements for schemas - Transaction blocks with BEGIN/COMMIT - Graph traversal queries Args: query_string: The complete SurrealQL query to execute. Examples: - "SELECT * FROM user WHERE age > 18" - "SELECT *, ->purchase->product FROM user:john" - "BEGIN; CREATE user:alice SET name = 'Alice'; CREATE user:bob SET name = 'Bob'; COMMIT;" 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 query executed successfully - data: The query results (format depends on the query) - error: Error message if the query failed (only present on failure) Example: >>> await query("SELECT name, age FROM user WHERE active = true ORDER BY created DESC LIMIT 10") {"success": true, "data": [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]} """ try: ns, db = resolve_namespace_database(namespace, database) logger.info(f"Executing query: {query_string[:100]}...") result = await repo_query(query_string, namespace=ns, database=db) # Format response return { "success": True, "data": result } except Exception as e: logger.error(f"Query failed: {str(e)}") raise Exception(f"SurrealDB query failed: {str(e)}")
• surreal_mcp/server.py:104-104 (registration)

  The @mcp.tool() decorator registers the query function as an MCP tool.
  @mcp.tool()
• surreal_mcp/database/__init__.py:64-91 (helper)

  Core helper function repo_query that performs the actual database query execution using pooled or override connections, 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 to resolve namespace and database from tool parameters or environment variables, determining whether to use pooled or override connections.
  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
• surreal_mcp/server.py:110-137 (schema)

  Docstring provides the input/output schema description for the 'query' tool, including parameters, returns, and examples.
  """ Execute a raw SurrealQL query against the connected SurrealDB database. This tool allows you to run any valid SurrealQL query directly. Use this for complex queries that don't fit the other tool patterns, such as: - Complex SELECT queries with JOINs, GROUP BY, or aggregations - Custom DEFINE statements for schemas - Transaction blocks with BEGIN/COMMIT - Graph traversal queries Args: query_string: The complete SurrealQL query to execute. Examples: - "SELECT * FROM user WHERE age > 18" - "SELECT *, ->purchase->product FROM user:john" - "BEGIN; CREATE user:alice SET name = 'Alice'; CREATE user:bob SET name = 'Bob'; COMMIT;" 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 query executed successfully - data: The query results (format depends on the query) - error: Error message if the query failed (only present on failure) Example: >>> await query("SELECT name, age FROM user WHERE active = true ORDER BY created DESC LIMIT 10") {"success": true, "data": [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]} """