On this page
MCP Bridge -- HTTP client that forwards MCP state to/from the collab server.
#server.src.pixelweaver.mcp_bridge
#server.src.pixelweaver.mcp_bridge
MCP Bridge -- HTTP client that forwards MCP state to/from the collab server.
Instead of the MCP server managing its own isolated state, every tool execution round-trips through the collaboration server's REST API:
1. Before execution: pull the latest state from the collab server 2. Execute the tool against the local (synced) state 3. After execution: push the modified state back, triggering a broadcast to all WebSocket clients (the frontend)
This ensures the collab server remains the single source of truth and that every MCP mutation is visible to the frontend in real time.
The collab server must be running on COLLAB_SERVER_URL for MCP to work.
#_get_client
def _get_client() -> httpx.AsyncClientReturn (and lazily create) the shared HTTP client.
#set_collab_server_url
def set_collab_server_url(url: str) -> NoneRetarget the bridge at a different collab server (called from CLI).
Updates the module-level URL and discards any lazily-created client so the next _get_client() rebinds to the new base URL. Without resetting the client, a client created before this call would keep talking to the old server -- silently mutating the wrong instance when two servers run.
#CollabServerUnreachableError
Raised when the collab server is not reachable.
#health_check
async def health_check() -> boolReturn True if the collab server is reachable.
#pull_full_state
async def pull_full_state() -> dict[str, Any]Fetch full serialized state from the collab server.
Returns the dict produced by GET /api/state/full. Raises CollabServerUnreachableError if the server cannot be reached.
#push_full_state
async def push_full_state(state_dict: dict[str, Any]) -> NonePush full serialized state to the collab server and trigger broadcast.
Sends POST /api/state/sync with the full state payload. Raises CollabServerUnreachableError if the server cannot be reached.
#serialize_state
def serialize_state(state: Any) -> dict[str, Any]Serialize a ServerState into a dict suitable for push_full_state.
Includes pixel data as base64-encoded strings so the JSON payload stays valid (raw bytes are not JSON-serializable).
#deserialize_into_state
def deserialize_into_state(state: Any, data: dict[str, Any]) -> NoneOverwrite a ServerState's contents from a serialized dict.
Inverse of serialize_state: restores projects, canvases, layers, pixel data, and history from the dict pulled from the collab server.