Aligns SDK endpoints with Vandoor telemetry architecture (#5)

TL;DR
-----

Aligns Python SDK with available or soon to be deployed endpoints

Details
--------

The Python SDK previously called several endpoints that don't exist in
Vandoor, causing instance tracking and metrics reporting to fail. This
change either 

This change adopts the telemetry-based architecture used by the Go
SDK. Instance metadata flows through telemetry headers to
`/kots_metrics/license_instance/info` rather than explicit CRUD
operations.

Authentication now correctly distinguishes between publishable keys
(which use Bearer prefix) and service tokens (raw token without
prefix), matching Vandoor's middleware expectations. Custom metrics
use the proper format with `{"data": {name: value}}` structure and
correct `/application/custom-metrics` endpoint.

Co-Authored-By: @jpshackelford
Co-Authored-By: Open Hands <noreply@all-hands.dev>
Co-Authored-by:  Claude Code <noreply@anthropic.com>

Author: 3 people

.envrc

@@ -0,0 +1,2 @@
+layout python python3.12
+dotenv_if_exists

.gitignore

@@ -23,14 +23,15 @@ share/python-wheels/
 *.egg
 MANIFEST
 
-# Virtual environments
+# Virtual environments 
 .env
 .venv
 env/
 venv/
 ENV/
 env.bak/
 venv.bak/
+.direnv
 
 # IDE
 .vscode/
@@ -61,4 +62,4 @@ Thumbs.db
 # Replicated SDK state
 Library/
 .local/
-AppData/
+AppData/

examples/async_example.py

@@ -5,7 +5,7 @@
 
 import asyncio
 
-from replicated import AsyncReplicatedClient, InstanceStatus
+from replicated import AsyncReplicatedClient
 
 
 async def main():
@@ -31,13 +31,6 @@ async def main():
         )
         print("Metrics sent successfully")
 
-        # Set the instance status and version concurrently
-        await asyncio.gather(
-            instance.set_status(InstanceStatus.RUNNING),
-            instance.set_version("1.2.0"),
-        )
-        print("Instance status set to RUNNING and version set to 1.2.0")
-
         print("Example completed successfully!")
 
 

examples/metrics_example.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""
+Basic example of using the Replicated Python SDK.
+This script initializes the replicated package, creates a customer and instance.
+"""
+
+import argparse
+import asyncio
+
+from replicated import AsyncReplicatedClient
+
+
+async def main():
+    parser = argparse.ArgumentParser(description="Basic Replicated SDK example")
+    parser.add_argument(
+        "--base-url",
+        default="https://replicated.app",
+        help="Base URL for the Replicated API (default: https://replicated.app)",
+    )
+    parser.add_argument(
+        "--publishable-key",
+        required=True,
+        help="Your Replicated publishable key (required)",
+    )
+    parser.add_argument(
+        "--app-slug", required=True, help="Your application slug (required)"
+    )
+    parser.add_argument(
+        "--customer-email",
+        default="user@example.com",
+        help="Customer email address (default: user@example.com)",
+    )
+    parser.add_argument("--channel", help="Channel for the customer (optional)")
+    parser.add_argument("--customer-name", help="Customer name (optional)")
+    parser.add_argument(
+        "--status",
+        choices=["missing", "unavailable", "ready", "updating", "degraded"],
+        default="ready",
+        help="Instance status (default: ready)",
+    )
+
+    args = parser.parse_args()
+
+    print("Initializing Replicated client...")
+    print(f"Base URL: {args.base_url}")
+    print(f"App Slug: {args.app_slug}")
+
+    # Initialize the client
+    async with AsyncReplicatedClient(
+        publishable_key=args.publishable_key,
+        app_slug=args.app_slug,
+        base_url=args.base_url,
+    ) as client:
+        print("✓ Replicated client initialized successfully")
+
+        # Create or get customer
+        channel_info = f" (channel: {args.channel})" if args.channel else ""
+        name_info = f" (name: {args.customer_name})" if args.customer_name else ""
+        print(
+            f"\nCreating/getting customer with email: "
+            f"{args.customer_email}{channel_info}{name_info}"
+        )
+        customer = await client.customer.get_or_create(
+            email_address=args.customer_email,
+            channel=args.channel,
+            name=args.customer_name,
+        )
+        print(f"✓ Customer created/retrieved - ID: {customer.customer_id}")
+
+        # Get or create the associated instance
+        instance = await customer.get_or_create_instance()
+        print(f"Instance ID: {instance.instance_id}")
+        print(f"✓ Instance created/retrieved - ID: {instance.instance_id}")
+
+        # Get or create the associated instance
+        instance = await customer.get_or_create_instance()
+        print(f"Instance ID: {instance.instance_id}")
+
+        # Set instance status
+        await instance.set_status(args.status)
+        print(f"✓ Instance status set to: {args.status}")
+
+        # Send some metrics concurrently
+        await asyncio.gather(
+            instance.send_metric("cpu_usage", 0.83),
+            instance.send_metric("memory_usage", 0.67),
+            instance.send_metric("disk_usage", 0.45),
+        )
+        print("Metrics sent successfully")
+
+    print(f"Instance ID: {instance.instance_id}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/sync_example.py

@@ -3,7 +3,7 @@
 Synchronous example of using the Replicated Python SDK.
 """
 
-from replicated import InstanceStatus, ReplicatedClient
+from replicated import ReplicatedClient
 
 
 def main():
@@ -27,14 +27,6 @@ def main():
         instance.send_metric("disk_usage", 0.45)
         print("Metrics sent successfully")
 
-        # Set the instance status
-        instance.set_status(InstanceStatus.RUNNING)
-        print("Instance status set to RUNNING")
-
-        # Set the application version
-        instance.set_version("1.2.0")
-        print("Instance version set to 1.2.0")
-
         print("Example completed successfully!")
 
 

replicated/__init__.py

@@ -9,7 +9,14 @@
     ReplicatedRateLimitError,
 )
 
-__version__ = "1.0.0"
+# Try to get version from package metadata, fall back to hardcoded version
+try:
+    from importlib.metadata import version as _get_version
+
+    __version__ = _get_version("replicated") + "+python"
+except Exception:
+    # Fallback for development or if package isn't installed
+    __version__ = "1.0.0+python"
 __all__ = [
     "ReplicatedClient",
     "AsyncReplicatedClient",

replicated/async_client.py

@@ -41,6 +41,8 @@ def _get_auth_headers(self) -> Dict[str, str]:
         # Try to use dynamic token first, fall back to publishable key
         dynamic_token = self.state_manager.get_dynamic_token()
         if dynamic_token:
-            return {"Authorization": f"Bearer {dynamic_token}"}
+            # Service tokens are sent without Bearer prefix
+            return {"Authorization": dynamic_token}
         else:
+            # Publishable keys use Bearer prefix
             return {"Authorization": f"Bearer {self.publishable_key}"}

replicated/client.py

@@ -41,6 +41,8 @@ def _get_auth_headers(self) -> Dict[str, str]:
         # Try to use dynamic token first, fall back to publishable key
         dynamic_token = self.state_manager.get_dynamic_token()
         if dynamic_token:
-            return {"Authorization": f"Bearer {dynamic_token}"}
+            # Service tokens are sent without Bearer prefix
+            return {"Authorization": dynamic_token}
         else:
+            # Publishable keys use Bearer prefix
             return {"Authorization": f"Bearer {self.publishable_key}"}

replicated/http_client.py

@@ -28,9 +28,14 @@ def _build_headers(
         self, headers: Optional[Dict[str, str]] = None
     ) -> Dict[str, str]:
         """Build request headers."""
+        # Import here to avoid circular import
+        from . import __version__
+
+        # Format: "Replicated-SDK/{version}" as expected by Vandoor
+        user_agent = f"Replicated-SDK/{__version__}"
         request_headers = {
             "Content-Type": "application/json",
-            "User-Agent": "replicated-python/1.0.0",
+            "User-Agent": user_agent,
             **self.default_headers,
         }
         if headers:
@@ -51,11 +56,6 @@ def _handle_response(self, response: httpx.Response) -> Dict[str, Any]:
         error_message = json_body.get("message", default_msg)
         error_code = json_body.get("code")
 
-        # Debug: print the full error response
-        print(f"DEBUG: HTTP {response.status_code} Error Response:")
-        print(f"DEBUG: Response body: {response.text}")
-        print(f"DEBUG: JSON body: {json_body}")
-
         if response.status_code == 401:
             raise ReplicatedAuthError(
                 message=error_message,