surreal-mcp

Instructions

Implementation Reference

• surreal_mcp/server.py:597-670 (handler)

  MCP tool handler for 'upsert'. Validates inputs, determines if new record or update, calls repo_upsert helper, and returns standardized response with success, data, and created flag.
  async def upsert( thing: str, data: Dict[str, Any], namespace: Optional[str] = None, database: Optional[str] = None, ) -> Dict[str, Any]: """ Upsert a record: create if it doesn't exist, merge/update if it does. This tool is perfect when you want to ensure a record exists with specific data, regardless of whether it already exists. It will: - Create a new record with the specified ID if it doesn't exist - Merge the provided data into the existing record if it does exist - Always succeed (unless there's a database error) Args: thing: The full record ID in format "table:id" (e.g., "user:john", "settings:global") data: The data for the record. If record exists, this will be merged with existing data 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 upsert was successful - data: The record after upserting - created: Boolean indicating if a new record was created (vs updated) - error: Error message if upsert failed (only present on failure) Examples: >>> await upsert("user:john", {"name": "John Doe", "email": "john@example.com"}) {"success": true, "data": {"id": "user:john", "name": "John Doe", ...}, "created": true} >>> await upsert("user:john", {"email": "newemail@example.com"}) # Update existing {"success": true, "data": {"id": "user:john", "name": "John Doe", "email": "newemail@example.com", ...}, "created": false} >>> await upsert("settings:global", {"theme": "dark", "language": "en"}) {"success": true, "data": {"id": "settings:global", "theme": "dark", "language": "en"}, "created": true} """ try: ns, db = resolve_namespace_database(namespace, database) # Validate thing format if ":" not in thing: raise ValueError(f"Invalid record ID format: {thing}. Must be 'table:id'") logger.info(f"Upserting record {thing}") # Check if record exists try: existing = await repo_query(f"SELECT * FROM {thing}", namespace=ns, database=db) created = not existing or len(existing) == 0 except Exception: created = True # Extract table name for repo_upsert table = thing.split(":", 1)[0] # Perform upsert - pass full record ID result = await repo_upsert( table=table, id=thing, data=data, add_timestamp=True, namespace=ns, database=db ) # Get the first result upserted_record = result[0] if result else {} return { "success": True, "data": upserted_record, "created": created } except Exception as e: logger.error(f"Upsert failed for {thing}: {str(e)}") raise Exception(f"Failed to upsert {thing}: {str(e)}")
• surreal_mcp/database/__init__.py:155-181 (helper)

  Core helper function implementing the SurrealQL UPSERT operation with MERGE, adding timestamps if requested, and executing via repo_query.
  async def repo_upsert( table: str, id: Optional[str], data: Dict[str, Any], add_timestamp: bool = False, namespace: Optional[str] = None, database: Optional[str] = None, ) -> List[Dict[str, Any]]: """Create or update a record in the specified table. Args: table: The table name id: Optional record ID (if provided, upserts that specific record) data: The record data to upsert add_timestamp: Whether to add/update the 'updated' timestamp namespace: Optional namespace override (uses env var if not provided) database: Optional database override (uses env var if not provided) Returns: The upserted record(s) """ data.pop("id", None) if add_timestamp: data["updated"] = datetime.now(timezone.utc) query = f"UPSERT {id if id else table} MERGE $data;" return await repo_query(query, {"data": data}, namespace=namespace, database=database)
• surreal_mcp/server.py:597-597 (registration)

  Decorator registering the upsert function as an MCP tool.
  async def upsert(
• surreal_mcp/server.py:55-99 (helper)

  Utility function to resolve namespace and database from tool parameters or environment variables, used by upsert and other tools.
  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