feat: add sign parameter to upload_data for notary-signed uploads (#44)

swarm_provenance_mcp/gateway_client.py

@@ -141,14 +141,16 @@ def extend_stamp(self, stamp_id: str, duration_hours: int) -> Dict[str, Any]:
         return response.json()
 
     def upload_data(
-        self, data: str, stamp_id: str, content_type: str = "application/json"
+        self, data: str, stamp_id: str, content_type: str = "application/json", sign: Optional[str] = None
     ) -> Dict[str, Any]:
         """Upload data to Swarm network.
 
         Args:
             data: Data content as string (max 4096 bytes)
             stamp_id: Postage stamp ID to use for upload
             content_type: MIME type of the content (default: application/json)
+            sign: Signing method for provenance. Use 'notary' to have the gateway
+                  cryptographically sign the data at upload time.
 
         Returns:
             Upload response with reference hash
@@ -170,6 +172,8 @@ def upload_data(
         files = {"file": ("data", data_bytes, content_type)}
 
         params = {"stamp_id": stamp_id, "content_type": content_type}
+        if sign:
+            params["sign"] = sign
 
         # For file uploads, temporarily remove Content-Type from session to let requests set multipart/form-data
         original_content_type = self.session.headers.pop("Content-Type", None)

swarm_provenance_mcp/server.py

@@ -582,6 +582,11 @@ async def list_tools() -> List[Tool]:
                             "description": "MIME type of the content (e.g., application/json, text/plain, image/png)",
                             "default": "application/json",
                         },
+                        "sign": {
+                            "type": "string",
+                            "description": "Signing method for provenance. Use 'notary' to have the gateway cryptographically sign the data at upload time, creating a verifiable proof of upload. Check notary_info first to confirm signing is available. Leave empty for unsigned upload.",
+                            "enum": ["notary"],
+                        },
                     },
                     "required": ["data", "stamp_id"],
                 },
@@ -1505,6 +1510,7 @@ async def handle_upload_data(arguments: Dict[str, Any]) -> CallToolResult:
         if not stamp_id:
             raise ValueError("Stamp ID cannot be empty")
         content_type = arguments.get("content_type", "application/json")
+        sign = arguments.get("sign")
 
         # Validate inputs
         validate_data_size(data)
@@ -1553,17 +1559,26 @@ async def handle_upload_data(arguments: Dict[str, Any]) -> CallToolResult:
                 raise
 
         # Proceed with upload if stamp validation passed
-        result = gateway_client.upload_data(data, clean_stamp_id, content_type)
+        result = gateway_client.upload_data(data, clean_stamp_id, content_type, sign=sign)
 
         response_text = f"🎉 Data uploaded successfully to Swarm!\n\n"
         response_text += f"📄 Upload Details:\n"
         response_text += f"   Size: {len(data.encode('utf-8')):,} bytes\n"
         response_text += f"   Content Type: {content_type}\n"
-        response_text += f"   Stamp Used: `{clean_stamp_id}`\n\n"
-        response_text += f"🔗 Retrieval Information:\n"
+        response_text += f"   Stamp Used: `{clean_stamp_id}`\n"
+        if sign:
+            response_text += f"   Signing: {sign}\n"
+        response_text += f"\n🔗 Retrieval Information:\n"
         response_text += f"   Reference Hash: `{result['reference']}`\n"
         response_text += f"   💡 Copy this reference hash to download your data later using the 'download_data' tool."
 
+        # Surface notary signature details when present
+        notary_info = result.get("notary") or result.get("signature")
+        if notary_info and isinstance(notary_info, dict):
+            signer = notary_info.get("signer") or notary_info.get("signerAddress")
+            if signer:
+                response_text += f"\n\n🔐 Notary Signature:\n   Signer: {signer}"
+
         # Add validation warning if applicable
         if stamp_validation_failed:
             response_text += f"\nNote: {validation_error_msg}"

tests/test_gateway_client.py

@@ -179,3 +179,31 @@ def test_custom_headers(self):
             assert "swarm-provenance-mcp" in request_headers["User-Agent"]
             assert "X-Payment-Mode" in request_headers
             assert request_headers["X-Payment-Mode"] == "free"
+
+    def test_upload_data_unsigned(self):
+        """Test that unsigned upload sends no sign query param."""
+        with requests_mock.Mocker() as m:
+            stamp_id = "a" * 64
+            expected_response = {"reference": "b" * 64}
+            m.post(f"{self.base_url}/api/v1/data/", json=expected_response)
+
+            result = self.client.upload_data('{"key": "value"}', stamp_id)
+
+            assert result["reference"] == "b" * 64
+            assert "sign" not in m.last_request.qs
+
+    def test_upload_data_with_notary_sign(self):
+        """Test that notary sign upload includes ?sign=notary query param."""
+        with requests_mock.Mocker() as m:
+            stamp_id = "a" * 64
+            signer_address = "0xabcdef1234567890abcdef1234567890abcdef12"
+            expected_response = {
+                "reference": "c" * 64,
+                "notary": {"signer": signer_address},
+            }
+            m.post(f"{self.base_url}/api/v1/data/", json=expected_response)
+
+            result = self.client.upload_data('{"key": "value"}', stamp_id, sign="notary")
+
+            assert result["reference"] == "c" * 64
+            assert m.last_request.qs.get("sign") == ["notary"]