Function streaming

skyvern/cli/doctor.py:1830–1865

Create or reuse a browser session and verify local browser streaming.

(
    base_url: str = typer.Option(
        "http://localhost:8000",
        "--base-url",
        help="Skyvern API server root or v1 URL.",
    ),
    api_key: str | None = typer.Option(None, "--api-key", help="Skyvern API key. Defaults to SKYVERN_API_KEY."),
    browser_session_id: str | None = typer.Option(
        None,
        "--browser-session-id",
        help="Existing browser session to test. If omitted, doctor creates and closes one.",
    ),
    timeout_seconds: int = typer.Option(45, "--timeout", min=5, max=180, help="Seconds to wait for stream readiness."),
    json_output: bool = typer.Option(False, "--json", help="Output machine-readable JSON."),
)

@doctor_app.command("streaming")
def streaming(
    base_url: str = typer.Option(
        "http://localhost:8000",
        "--base-url",
        help="Skyvern API server root or v1 URL.",
    ),
    api_key: str | None = typer.Option(None, "--api-key", help="Skyvern API key. Defaults to SKYVERN_API_KEY."),
    browser_session_id: str | None = typer.Option(
        None,
        "--browser-session-id",
        help="Existing browser session to test. If omitted, doctor creates and closes one.",
    ),
    timeout_seconds: int = typer.Option(45, "--timeout", min=5, max=180, help="Seconds to wait for stream readiness."),
    json_output: bool = typer.Option(False, "--json", help="Output machine-readable JSON."),
) -> None:
    """Create or reuse a browser session and verify local browser streaming."""
    result = _streaming_smoke_test(
        base_url=base_url,
        api_key=api_key,
        browser_session_id=browser_session_id,
        timeout_seconds=timeout_seconds,
    )

    if json_output:
        console.print(json.dumps(result.__dict__))
    else:
        table = Table(show_header=True, header_style="bold", title="Skyvern Streaming Doctor")
        table.add_column("Check", style="bold")
        table.add_column("Status")
        table.add_column("Details")
        table.add_column("Fix")
        table.add_row(result.name, _STATUS_STYLE.get(result.status, result.status), result.detail, result.hint)
        console.print(table)

    if result.status == "error":
        raise typer.Exit(code=1)


@doctor_app.callback(invoke_without_command=True)