Skip to content

sim2bot.ensure_bridge

ensure_bridge(url: str = DEFAULT_CONTROL_URL, *, host: Optional[str] = None, tcp_port: int = DEFAULT_TCP_PORT, udp_port: int = DEFAULT_UDP_PORT, timeout: float = 10.0) -> BridgeInfo

Reuse a healthy local bridge or start one in the background.

Parameters:

  • url (str, default: DEFAULT_CONTROL_URL ) –

    Desired control WebSocket URL.

  • host (Optional[str], default: None ) –

    Bind host for a new process. Defaults to loopback unless configured with BRIDGE_HOST.

  • tcp_port (int, default: DEFAULT_TCP_PORT ) –

    Newline-delimited JSON TCP port.

  • udp_port (int, default: DEFAULT_UDP_PORT ) –

    Datagram control/telemetry port.

  • timeout (float, default: 10.0 ) –

    Seconds to wait for a new bridge to become healthy.

Returns:

Raises:

  • RuntimeError –

    If a new bridge exits early or does not become healthy.

Notes

The subprocess uses the current Python interpreter and is terminated at interpreter exit. Use the sim2bot bridge CLI for a manually managed, long-running process. The default loopback bind is deliberate; LAN access requires explicit host and authentication configuration.

Source code in sim2bot/bridge.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
def ensure_bridge(
    url: str = DEFAULT_CONTROL_URL,
    *,
    host: Optional[str] = None,
    tcp_port: int = DEFAULT_TCP_PORT,
    udp_port: int = DEFAULT_UDP_PORT,
    timeout: float = 10.0,
) -> BridgeInfo:
    """Reuse a healthy local bridge or start one in the background.

    Args:
        url: Desired control WebSocket URL.
        host: Bind host for a new process. Defaults to loopback unless configured
            with ``BRIDGE_HOST``.
        tcp_port: Newline-delimited JSON TCP port.
        udp_port: Datagram control/telemetry port.
        timeout: Seconds to wait for a new bridge to become healthy.

    Returns:
        Connection and ownership details as
        [`BridgeInfo`][sim2bot.bridge.BridgeInfo].

    Raises:
        RuntimeError: If a new bridge exits early or does not become healthy.

    Notes:
        The subprocess uses the current Python interpreter and is terminated at
        interpreter exit. Use the ``sim2bot bridge`` CLI for a manually managed,
        long-running process. The default loopback bind is deliberate; LAN access
        requires explicit host and authentication configuration.
    """
    if is_bridge_running(url):
        return bridge_info(url, tcp_port=tcp_port, udp_port=udp_port)

    parsed = urlparse(url)
    bind_host = host or os.environ.get("BRIDGE_HOST") or "127.0.0.1"
    port = parsed.port or 8765
    env = os.environ.copy()
    env["BRIDGE_TCP_PORT"] = str(tcp_port)
    env["BRIDGE_UDP_PORT"] = str(udp_port)
    env["BRIDGE_RAW_HOST"] = bind_host
    env.setdefault("PYTHONUNBUFFERED", "1")

    cmd = [
        sys.executable,
        "-m",
        "uvicorn",
        "sim2bot.bridge_server:app",
        "--host",
        bind_host,
        "--port",
        str(port),
        "--log-level",
        "warning",
    ]
    process = subprocess.Popen(
        cmd,
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        start_new_session=True,
    )
    _managed_processes.append(process)
    _register_cleanup()

    if wait_for_bridge(url, timeout=timeout):
        return bridge_info(
            url,
            tcp_port=tcp_port,
            udp_port=udp_port,
            started_by_sdk=True,
            pid=process.pid,
        )

    exit_code = process.poll()
    if is_bridge_running(url):
        return bridge_info(url, tcp_port=tcp_port, udp_port=udp_port)
    if exit_code is None:
        _terminate_process(process)
        raise RuntimeError(f"Sim2Bot bridge did not become healthy within {timeout:.1f}s")
    raise RuntimeError(
        f"Sim2Bot bridge failed to start (exit code {exit_code}); "
        "run `sim2bot bridge` to see server logs"
    )

Practical examples

Examples use objects discovered from the current scene rather than hard-coded robot metadata.

Start or reuse the default bridge

from sim2bot import ensure_bridge

info = ensure_bridge()
print(info.health_url)

Choose explicit local ports

info = ensure_bridge(
    url="ws://127.0.0.1:9000/ws",
    tcp_port=9001,
    udp_port=9002,
    timeout=15.0,
)

Guidance

Usage tip

Use Robot(auto_bridge=True) unless your application specifically needs bridge process details.

Important behavior

The default bind is loopback. Exposing the bridge to a LAN requires deliberate authentication and firewall configuration.

See also