{"id":7247,"library":"foxglove-sdk","title":"Foxglove Python SDK","description":"The Foxglove Python SDK provides tools for streaming data (e.g., sensor data, robot state, custom messages) from Python applications to Foxglove Studio for visualization and debugging. It's actively maintained with regular releases, currently at version 0.21.0, targeting Python 3.10 and newer, leveraging asynchronous operations for efficient data transfer.","status":"active","version":"0.21.0","language":"en","source_language":"en","source_url":"https://github.com/foxglove/foxglove-sdk-py","tags":["robotics","data streaming","visualization","asyncio","websocket"],"install":[{"cmd":"pip install foxglove-sdk","lang":"bash","label":"Install stable version"}],"dependencies":[{"reason":"Data validation and settings management","package":"pydantic","optional":false},{"reason":"Asynchronous WebSocket client implementation","package":"websockets","optional":false},{"reason":"HTTP client for schema uploads (e.g., Protobuf)","package":"httpx","optional":false}],"imports":[{"note":"The primary client class was renamed from `Client` to `FoxgloveClient` in v0.20.0 for clarity and direct top-level access.","wrong":"from foxglove_sdk.client import Client","symbol":"FoxgloveClient","correct":"from foxglove_sdk import FoxgloveClient"},{"note":"In v0.18.0, the concept of a 'Channel' was renamed to 'Stream' throughout the API.","wrong":"from foxglove_sdk.client import AddChannelRequest","symbol":"AddStreamRequest","correct":"from foxglove_sdk.client import AddStreamRequest"},{"symbol":"MessageEvent","correct":"from foxglove_sdk.base import MessageEvent"}],"quickstart":{"code":"import asyncio\nimport os\nimport time\nimport json\nfrom foxglove_sdk import FoxgloveClient\nfrom foxglove_sdk.client import AddStreamRequest, ClientError\nfrom foxglove_sdk.base import MessageEvent\n\nasync def main():\n    # Connect to a local Foxglove WebSocket server (e.g., `npx @foxglove/ws-server`)\n    # or a Foxglove Cloud stream endpoint.\n    websocket_url = os.environ.get(\"FOXGLOVE_WEBSOCKET_URL\", \"ws://localhost:8765/\")\n    print(f\"Attempting to connect to Foxglove WebSocket at {websocket_url}\")\n\n    client = FoxgloveClient(websocket_url)\n    try:\n        await client.connect()\n        print(\"Successfully connected to Foxglove WebSocket.\")\n\n        # Define a simple JSON schema for a counter stream\n        stream_id = await client.add_stream(\n            AddStreamRequest(\n                topic=\"my_counter\",\n                encoding=\"json\",\n                schemaName=\"example.Counter\",\n                schema=json.dumps({\n                    \"type\": \"object\",\n                    \"properties\": {\n                        \"count\": {\"type\": \"integer\"},\n                        \"timestamp\": {\"type\": \"number\"}\n                    },\n                    \"required\": [\"count\", \"timestamp\"]\n                })\n            )\n        )\n        print(f\"Stream 'my_counter' (ID: {stream_id}) added.\")\n\n        # Publish 10 counter messages\n        for i in range(10):\n            current_time_ns = time.time_ns() # Timestamp in nanoseconds\n            message_payload = {\"count\": i, \"timestamp\": current_time_ns / 1e9}\n            await client.send_message(\n                MessageEvent(\n                    timestamp=current_time_ns,\n                    streamId=stream_id,\n                    payload=json.dumps(message_payload).encode(\"utf8\") # Payload must be bytes\n                )\n            )\n            print(f\"Published message {i}: {message_payload}\")\n            await asyncio.sleep(1) # Wait for 1 second before next message\n\n    except ClientError as e:\n        print(f\"Foxglove Client Error: {e}\")\n        print(\"Please ensure a Foxglove WebSocket server is running and accessible (e.g., `npx @foxglove/ws-server`).\")\n    except Exception as e:\n        print(f\"An unexpected error occurred: {e}\")\n    finally:\n        if client.connected:\n            await client.disconnect()\n            print(\"Disconnected from Foxglove WebSocket.\")\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n","lang":"python","description":"This quickstart demonstrates how to connect to a Foxglove WebSocket server, define a new stream with a JSON schema, and publish messages asynchronously. Ensure you have a Foxglove WebSocket server running (e.g., locally via `npx @foxglove/ws-server` if you have Node.js installed, or connect to a cloud endpoint)."},"warnings":[{"fix":"Replace `from foxglove_sdk.client import Client` with `from foxglove_sdk import FoxgloveClient`. Instead of `client.publisher.send()`, use `client.send_message()` directly.","message":"The `Publisher` class was removed and the main client class was renamed from `Client` to `FoxgloveClient` in version 0.20.0. The API was simplified for direct message sending.","severity":"breaking","affected_versions":"0.20.0+"},{"fix":"Update all references from 'channel' to 'stream' in class names, method calls, and parameter names (e.g., `add_channel` -> `add_stream`, `channelId` -> `streamId`).","message":"The term 'Channel' was renamed to 'Stream' across the entire API in version 0.18.0. This affects class names (e.g., `AddChannelRequest` to `AddStreamRequest`) and method parameters.","severity":"breaking","affected_versions":"0.18.0+"},{"fix":"Ensure your application uses an `asyncio` event loop. Wrap the main logic in an `async def main():` function and run it with `asyncio.run(main())`. Always use `await` when calling `FoxgloveClient` methods like `connect()`, `add_stream()`, or `send_message()`.","message":"The Foxglove SDK is built on `asyncio`. All network operations are asynchronous and must be `await`ed within an `async` function. Running synchronous code or attempting to call async methods without `await` will result in `RuntimeError` or unexpected behavior.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Before sending, serialize your message object (e.g., `json.dumps(my_dict).encode('utf8')` for JSON, or `my_proto_message.SerializeToString()` for Protobuf) and ensure the schema and encoding in `AddStreamRequest` match this format.","message":"Data payloads sent via `send_message` must be bytes, not raw Python dicts or strings. The schema defined during `add_stream` must precisely match the structure of the data you send.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"fix":"The `Publisher` class was removed. Directly use `client.send_message()` after creating a `FoxgloveClient` instance. Remove the `Publisher` import and any `client.publisher` calls.","cause":"Attempting to import the `Publisher` class, which was removed in `foxglove-sdk` v0.20.0.","error":"ImportError: cannot import name 'Publisher' from 'foxglove_sdk'"},{"fix":"Update the import statement to `from foxglove_sdk import FoxgloveClient`.","cause":"Attempting to import the `Client` class, which was renamed to `FoxgloveClient` and moved to the top-level package in v0.20.0.","error":"ImportError: cannot import name 'Client' from 'foxglove_sdk.client'"},{"fix":"If running in an interactive environment or existing event loop, get the current loop and create tasks directly, e.g., `loop = asyncio.get_event_loop(); loop.create_task(main())`. Avoid calling `asyncio.run()` more than once in the same thread, or ensure it's only called when no loop is running.","cause":"Attempting to run `asyncio.run()` when an `asyncio` event loop is already active in the current thread (e.g., in a Jupyter notebook or another async context).","error":"RuntimeError: Event loop is already running"},{"fix":"Carefully review the `schemaName` and `schema` provided in your `AddStreamRequest`. Ensure the `schema` is valid JSON (or Protobuf schema string) and correctly represents the data you intend to send. Check server logs for more specific errors.","cause":"The WebSocket server rejected the stream registration, often due to an invalid or malformed schema, or a topic name conflict.","error":"foxglove_sdk.client.exceptions.ClientError: Failed to register stream"}]}