surreal-mcp

Instructions

Implementation Reference

• surreal_mcp/server.py:534-631 (handler)

  The 'patch' tool handler: an async function decorated with @mcp.tool() that implements JSON Patch operations by converting them to a SurrealDB MERGE via repo_upsert. Handles validation, path conversion, and returns the updated record.

  @mcp.tool()
  async def patch(
      thing: str,
      patches: List[Dict[str, Any]],
      namespace: Optional[str] = None,
      database: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Apply JSON Patch operations to a specific record (RFC 6902).
  
      This tool applies a sequence of patch operations to modify a record. However, since SurrealDB
      doesn't natively support JSON Patch, this implementation converts patches to a merge operation.
      Supported operations:
      - add: Add a new field or array element
      - remove: Remove a field (limited support)
      - replace: Replace a field value
  
      Args:
          thing: The full record ID to patch in format "table:id" (e.g., "user:john")
          patches: Array of patch operations. Each operation should have:
              - op: The operation type ("add", "remove", "replace", "move", "copy", "test")
              - path: The field path (e.g., "/email", "/profile/bio")
              - value: The value for add/replace operations
          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 patch was successful
          - data: The complete record after applying patches
          - applied_patches: Number of patch operations applied
          - error: Error message if patch failed (only present on failure)
  
      Examples:
          >>> await patch("user:john", [
          ...     {"op": "replace", "path": "/email", "value": "john@newdomain.com"},
          ...     {"op": "add", "path": "/verified", "value": true}
          ... ])
          {
              "success": true,
              "data": {"id": "user:john", "email": "john@newdomain.com", "verified": true, ...},
              "applied_patches": 2
          }
  
      Note: This provides compatibility with JSON Patch but internally uses SurrealDB's merge.
      Complex operations like "move" or "test" are not fully supported.
      """
      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'")
  
          if not patches or not isinstance(patches, list):
              raise ValueError("Patches must be a non-empty array")
  
          logger.info(f"Applying {len(patches)} patches to {thing}")
  
          # Convert JSON Patch operations to a merge object
          merge_data = {}
          for patch_op in patches:
              op = patch_op.get("op")
              path = patch_op.get("path", "")
              value = patch_op.get("value")
  
              # Remove leading slash and convert path to field name
              field = path.lstrip("/").replace("/", ".")
  
              if op in ["add", "replace"]:
                  merge_data[field] = value
              elif op == "remove":
                  # Note: SurrealDB doesn't support removing fields via MERGE
                  # This would need a custom UPDATE query
                  logger.warning(f"Remove operation on {field} not fully supported")
              else:
                  logger.warning(f"Patch operation '{op}' not supported")
  
          # Extract table name for repo_upsert
          table = thing.split(":", 1)[0]
  
          # Apply the patches via merge - pass full record ID
          result = await repo_upsert(
              table=table, id=thing, data=merge_data, add_timestamp=True, namespace=ns, database=db
          )
  
          # Get the first result
          patched_record = result[0] if result else {}
  
          return {
              "success": True,
              "data": patched_record,
              "applied_patches": len(patches)
          }
      except Exception as e:
          logger.error(f"Patch failed for {thing}: {str(e)}")
          raise Exception(f"Failed to patch {thing}: {str(e)}")