Foxglove Python SDK

Common errors

• ImportError: cannot import name 'Publisher' from 'foxglove_sdk'

  cause Attempting to import the `Publisher` class, which was removed in `foxglove-sdk` v0.20.0.
  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.

• ImportError: cannot import name 'Client' from 'foxglove_sdk.client'

  cause Attempting to import the `Client` class, which was renamed to `FoxgloveClient` and moved to the top-level package in v0.20.0.
  fix Update the import statement to `from foxglove_sdk import FoxgloveClient`.

• RuntimeError: Event loop is already 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).
  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.

• foxglove_sdk.client.exceptions.ClientError: Failed to register stream

  cause The WebSocket server rejected the stream registration, often due to an invalid or malformed schema, or a topic name conflict.
  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.

Warnings

• breaking 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.
  Fix: Replace `from foxglove_sdk.client import Client` with `from foxglove_sdk import FoxgloveClient`. Instead of `client.publisher.send()`, use `client.send_message()` directly.
• breaking 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.
  Fix: Update all references from 'channel' to 'stream' in class names, method calls, and parameter names (e.g., `add_channel` -> `add_stream`, `channelId` -> `streamId`).
• gotcha 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.
  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()`.
• gotcha 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.
  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.

Install

• pip install foxglove-sdk Install stable version

Imports

• FoxgloveClient
  wrong:

  from foxglove_sdk.client import Client

  correct:

  from foxglove_sdk import FoxgloveClient

  The primary client class was renamed from `Client` to `FoxgloveClient` in v0.20.0 for clarity and direct top-level access.
• AddStreamRequest
  wrong:

  from foxglove_sdk.client import AddChannelRequest

  correct:

  from foxglove_sdk.client import AddStreamRequest

  In v0.18.0, the concept of a 'Channel' was renamed to 'Stream' throughout the API.
• MessageEvent

  from foxglove_sdk.base import MessageEvent

Quickstart

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).

import asyncio
import os
import time
import json
from foxglove_sdk import FoxgloveClient
from foxglove_sdk.client import AddStreamRequest, ClientError
from foxglove_sdk.base import MessageEvent

async def main():
    # Connect to a local Foxglove WebSocket server (e.g., `npx @foxglove/ws-server`)
    # or a Foxglove Cloud stream endpoint.
    websocket_url = os.environ.get("FOXGLOVE_WEBSOCKET_URL", "ws://localhost:8765/")
    print(f"Attempting to connect to Foxglove WebSocket at {websocket_url}")

    client = FoxgloveClient(websocket_url)
    try:
        await client.connect()
        print("Successfully connected to Foxglove WebSocket.")

        # Define a simple JSON schema for a counter stream
        stream_id = await client.add_stream(
            AddStreamRequest(
                topic="my_counter",
                encoding="json",
                schemaName="example.Counter",
                schema=json.dumps({
                    "type": "object",
                    "properties": {
                        "count": {"type": "integer"},
                        "timestamp": {"type": "number"}
                    },
                    "required": ["count", "timestamp"]
                })
            )
        )
        print(f"Stream 'my_counter' (ID: {stream_id}) added.")

        # Publish 10 counter messages
        for i in range(10):
            current_time_ns = time.time_ns() # Timestamp in nanoseconds
            message_payload = {"count": i, "timestamp": current_time_ns / 1e9}
            await client.send_message(
                MessageEvent(
                    timestamp=current_time_ns,
                    streamId=stream_id,
                    payload=json.dumps(message_payload).encode("utf8") # Payload must be bytes
                )
            )
            print(f"Published message {i}: {message_payload}")
            await asyncio.sleep(1) # Wait for 1 second before next message

    except ClientError as e:
        print(f"Foxglove Client Error: {e}")
        print("Please ensure a Foxglove WebSocket server is running and accessible (e.g., `npx @foxglove/ws-server`).")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
    finally:
        if client.connected:
            await client.disconnect()
            print("Disconnected from Foxglove WebSocket.")

if __name__ == "__main__":
    asyncio.run(main())