surreal-mcp

Instructions

Implementation Reference

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

  Main handler for the 'query' tool: executes multiple SurrealQL queries, handles errors per query, resolves ns/db, calls repo_query helper.
  @mcp.tool() async def query( queries: List[str], namespace: Optional[str] = None, database: Optional[str] = None, ) -> Dict[str, Any]: """ Execute one or more SurrealQL queries against the connected SurrealDB database. This tool allows you to run any valid SurrealQL queries 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 Queries are executed sequentially. If a query fails, execution continues with the remaining queries, and the error is captured in that query's result. Args: queries: A list of SurrealQL queries to execute. Examples: - ["SELECT * FROM user WHERE age > 18"] - ["SELECT * FROM user", "SELECT * FROM product"] - ["CREATE user:alice SET name = 'Alice'", "CREATE user:bob SET name = 'Bob'"] 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 at least one query executed successfully - results: Array of per-query results, each containing: - success: Boolean indicating if this specific query succeeded - data: The query results (only present on success) - error: Error message (only present on failure) - total: Total number of queries executed - succeeded: Number of queries that succeeded - failed: Number of queries that failed Example: >>> await query(["SELECT * FROM user", "SELECT * FROM product"]) { "success": true, "results": [ {"success": true, "data": [{"id": "user:1", "name": "Alice"}]}, {"success": true, "data": [{"id": "product:1", "name": "Laptop"}]} ], "total": 2, "succeeded": 2, "failed": 0 } """ if not queries or not isinstance(queries, list): raise ValueError("queries must be a non-empty list of query strings") ns, db = resolve_namespace_database(namespace, database) results = [] succeeded = 0 failed = 0 for query_string in queries: try: logger.info(f"Executing query: {query_string[:100]}...") result = await repo_query(query_string, namespace=ns, database=db) results.append({ "success": True, "data": result }) succeeded += 1 except Exception as e: logger.error(f"Query failed: {str(e)}") results.append({ "success": False, "error": f"SurrealDB query failed: {str(e)}" }) failed += 1 return { "success": succeeded > 0, "results": results, "total": len(queries), "succeeded": succeeded, "failed": failed }
• surreal_mcp/database/__init__.py:64-91 (helper)

  Helper function that executes the actual SurrealQL query using pooled or override DB connection, 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-98 (helper)

  Helper function to resolve namespace and database from tool params or env vars, deciding pooled vs override connection.
  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:104-104 (registration)

  Registration of the 'query' tool using FastMCP @tool decorator.
  @mcp.tool()
• surreal_mcp/server.py:110-154 (schema)

  Input/output schema and documentation for the 'query' tool defined in the docstring and type hints.
  """ Execute one or more SurrealQL queries against the connected SurrealDB database. This tool allows you to run any valid SurrealQL queries 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 Queries are executed sequentially. If a query fails, execution continues with the remaining queries, and the error is captured in that query's result. Args: queries: A list of SurrealQL queries to execute. Examples: - ["SELECT * FROM user WHERE age > 18"] - ["SELECT * FROM user", "SELECT * FROM product"] - ["CREATE user:alice SET name = 'Alice'", "CREATE user:bob SET name = 'Bob'"] 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 at least one query executed successfully - results: Array of per-query results, each containing: - success: Boolean indicating if this specific query succeeded - data: The query results (only present on success) - error: Error message (only present on failure) - total: Total number of queries executed - succeeded: Number of queries that succeeded - failed: Number of queries that failed Example: >>> await query(["SELECT * FROM user", "SELECT * FROM product"]) { "success": true, "results": [ {"success": true, "data": [{"id": "user:1", "name": "Alice"}]}, {"success": true, "data": [{"id": "product:1", "name": "Laptop"}]} ], "total": 2, "succeeded": 2, "failed": 0 } """