Add command-aware env scope resolver (#5928)
Some checks are pending
Auto Create GitHub Release on Version Change / check-version-change (push) Waiting to run
Auto Create GitHub Release on Version Change / create-release (push) Blocked by required conditions
Run tests and pre-commit / Run tests and pre-commit hooks (push) Waiting to run
Run tests and pre-commit / Frontend Lint and Build (push) Waiting to run
Publish Fern Docs / run (push) Waiting to run
Build Skyvern SDK and publish to PyPI / check-version-change (push) Waiting to run
Build Skyvern SDK and publish to PyPI / run-ci (push) Blocked by required conditions
Build Skyvern SDK and publish to PyPI / build-sdk (push) Blocked by required conditions

This commit is contained in:
Marc Kelechava 2026-05-09 20:15:32 -07:00 committed by GitHub
parent 0b38970794
commit 10c56e4895
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 1355 additions and 189 deletions

View file

@ -1,9 +1,15 @@
import logging
from collections.abc import Set
import os
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from skyvern.utils.env_paths import EnvIntent
_DEFAULT_CLI_LOG_LEVEL = "WARNING"
_QUIET_CLI_LOGGERS = ("skyvern", "httpx", "litellm", "playwright", "httpcore")
_RUNTIME_LOGGING_CONFIGURED = False
def raise_unless_missing_optional_dependency(exc: ImportError, expected_modules: set[str]) -> None:
@ -12,27 +18,39 @@ def raise_unless_missing_optional_dependency(exc: ImportError, expected_modules:
raise exc
def _resolve_cli_log_level_name(settings: object) -> str:
"""Honor explicit settings while keeping CLI defaults quiet."""
fields_set: Set[str] = getattr(settings, "model_fields_set", set())
configured_level = str(getattr(settings, "LOG_LEVEL", _DEFAULT_CLI_LOG_LEVEL)).upper()
if "LOG_LEVEL" in fields_set:
return configured_level
return _DEFAULT_CLI_LOG_LEVEL
def _resolve_cli_log_level_name() -> str:
"""Honor explicit process env while keeping CLI defaults quiet."""
return os.environ.get("LOG_LEVEL", _DEFAULT_CLI_LOG_LEVEL).upper()
def configure_cli_bootstrap_logging() -> None:
"""Clamp CLI process logging before importing the command tree."""
from skyvern.config import settings # noqa: PLC0415
from skyvern.forge.sdk.forge_log import setup_logger # noqa: PLC0415
log_level_name = _resolve_cli_log_level_name(settings)
settings.LOG_LEVEL = log_level_name
setup_logger()
log_level_name = _resolve_cli_log_level_name()
log_level = logging.getLevelName(log_level_name)
if not isinstance(log_level, int):
log_level = logging.WARNING
logging.getLogger().setLevel(log_level)
for logger_name in _QUIET_CLI_LOGGERS:
logging.getLogger(logger_name).setLevel(log_level)
def configure_cli_runtime_logging() -> None:
"""Configure full Skyvern logging after CLI env intent has been selected."""
global _RUNTIME_LOGGING_CONFIGURED
if _RUNTIME_LOGGING_CONFIGURED:
return
from skyvern.forge.sdk.forge_log import setup_logger # noqa: PLC0415
setup_logger()
_RUNTIME_LOGGING_CONFIGURED = True
def prepare_cli_runtime(intent: "EnvIntent | str") -> Path:
"""Load intent-scoped env files, then configure logging that imports settings."""
from skyvern.utils.env_paths import load_backend_env_files # noqa: PLC0415
env_path = load_backend_env_files(intent=intent)
configure_cli_runtime_logging()
return env_path

View file

@ -6,6 +6,7 @@ import socket
import threading
import urllib.parse
import webbrowser
from pathlib import Path
import typer
@ -109,11 +110,14 @@ class _CallbackHandler(http.server.BaseHTTPRequestHandler):
def run_signup(
base_url: str = _DEFAULT_FRONTEND_URL,
timeout: int = _CALLBACK_TIMEOUT,
env_path: Path | str | None = None,
) -> str | None:
"""Core signup logic. Called by both the Typer command and init_command.
Returns the API key on success, or None if signup failed before saving.
"""
from skyvern.utils.env_paths import EnvIntent, resolve_backend_env_path
from .llm_setup import update_or_add_env_var
bound_socket = _find_free_port()
@ -153,15 +157,20 @@ def run_signup(
api_base_url = _derive_api_base_url(base_url)
update_or_add_env_var("SKYVERN_API_KEY", api_key)
update_or_add_env_var("SKYVERN_BASE_URL", api_base_url)
saved_env_path = (
Path(env_path).expanduser()
if env_path is not None
else resolve_backend_env_path(intent=EnvIntent.CLOUD, for_write=True)
)
update_or_add_env_var("SKYVERN_API_KEY", api_key, env_path=saved_env_path)
update_or_add_env_var("SKYVERN_BASE_URL", api_base_url, env_path=saved_env_path)
console.print("\n[bold green]Signup successful![/bold green]")
if email:
console.print(f"Email: {email}")
if organization_id:
console.print(f"Organization: {organization_id}")
console.print("API key saved to .env")
console.print(f"API key saved to {saved_env_path}")
console.print(f"Base URL: {api_base_url}")
return api_key

View file

@ -5,18 +5,27 @@ from __future__ import annotations
from typing import Any
import typer
from dotenv import load_dotenv
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.utils.env_paths import EnvIntent
from .commands._output import resolve_inline_or_file, run_tool
from .mcp_tools.blocks import skyvern_block_schema as tool_block_schema
from .mcp_tools.blocks import skyvern_block_validate as tool_block_validate
block_app = typer.Typer(help="Workflow block schema and validation commands.", no_args_is_help=True)
async def tool_block_schema(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.blocks import skyvern_block_schema # noqa: PLC0415
return await skyvern_block_schema(**kwargs)
async def tool_block_validate(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.blocks import skyvern_block_validate # noqa: PLC0415
return await skyvern_block_validate(**kwargs)
@block_app.callback()
def block_callback(
api_key: str | None = typer.Option(
@ -27,8 +36,10 @@ def block_callback(
),
) -> None:
"""Load environment and optional API key override."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
if api_key:
from skyvern.config import settings # noqa: PLC0415
settings.SKYVERN_API_KEY = api_key

View file

@ -1,15 +1,13 @@
import json
import platform
import time
from typing import Optional
from typing import Any, Optional
from urllib.parse import urlparse
import requests # type: ignore
from rich.panel import Panel
from rich.prompt import Confirm, Prompt
from skyvern.analytics import capture_setup_event
from .console import console
from .core.browser_launcher import (
SKYVERN_DATA_DIR,
@ -21,6 +19,18 @@ from .core.browser_launcher import (
_CDP_SCAN_PORTS = [9222, 9223, 9224, 9225, 9226, 9229]
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(event_name, success, error_type, error_message, extra_data)
def _check_cdp_ws(port: int) -> Optional[dict]:
"""Try a WebSocket CDP handshake on the given port.

View file

@ -1,9 +1,7 @@
import typer
from dotenv import load_dotenv
from skyvern._cli_bootstrap import configure_cli_bootstrap_logging as _configure_cli_bootstrap_logging
from skyvern.cli.lazy import LazyTyperGroup, register_lazy_command
from skyvern.utils.env_paths import resolve_backend_env_path
_cli_logging_configured = False
@ -241,5 +239,4 @@ def _walk_command_tree(cmd: object, prefix: str = "", max_depth: int = 1, _curre
if __name__ == "__main__": # pragma: no cover - manual CLI invocation
load_dotenv(resolve_backend_env_path())
cli_app()

View file

@ -1,15 +1,10 @@
from dotenv import load_dotenv
from skyvern._cli_bootstrap import configure_cli_bootstrap_logging, raise_unless_missing_optional_dependency
from skyvern.utils.env_paths import resolve_backend_env_path
def main() -> None:
configure_cli_bootstrap_logging()
from . import cli_app # noqa: PLC0415
load_dotenv(resolve_backend_env_path())
try:
from skyvern.cli.core.telemetry import register_cli_telemetry_flush # noqa: PLC0415
except ImportError as exc:

View file

@ -5,15 +5,12 @@ from __future__ import annotations
from typing import Any
import typer
from dotenv import load_dotenv
from skyvern.config import settings
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.forge.sdk.schemas.organizations import Organization, OrganizationUpdate
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import EnvIntent
from .commands._output import run_tool
from .mcp_tools.org import skyvern_org_get as tool_org_get
from .mcp_tools.org import skyvern_org_update as tool_org_update
_SETTABLE_KEYS: frozenset[str] = frozenset(OrganizationUpdate.model_fields)
# ``clear_*`` keys are write-only verbs (reset to NULL); not surfaced by ``get``.
@ -28,6 +25,18 @@ config_app = typer.Typer(
)
async def tool_org_get() -> dict[str, Any]:
from .mcp_tools.org import skyvern_org_get # noqa: PLC0415
return await skyvern_org_get()
async def tool_org_update(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.org import skyvern_org_update # noqa: PLC0415
return await skyvern_org_update(**kwargs)
@config_app.callback()
def config_callback(
api_key: str | None = typer.Option(
@ -38,8 +47,10 @@ def config_callback(
),
) -> None:
"""Load env and apply the optional API key override."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
if api_key:
from skyvern.config import settings # noqa: PLC0415
settings.SKYVERN_API_KEY = api_key

View file

@ -5,15 +5,11 @@ from __future__ import annotations
from typing import Any
import typer
from dotenv import load_dotenv
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.utils.env_paths import EnvIntent
from .commands._output import run_tool
from .mcp_tools.credential import skyvern_credential_delete as tool_credential_delete
from .mcp_tools.credential import skyvern_credential_get as tool_credential_get
from .mcp_tools.credential import skyvern_credential_list as tool_credential_list
credential_app = typer.Typer(
help="MCP-parity credential commands (list/get/delete). Use `skyvern credentials add` for secure creation.",
@ -21,6 +17,24 @@ credential_app = typer.Typer(
)
async def tool_credential_list(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.credential import skyvern_credential_list # noqa: PLC0415
return await skyvern_credential_list(**kwargs)
async def tool_credential_get(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.credential import skyvern_credential_get # noqa: PLC0415
return await skyvern_credential_get(**kwargs)
async def tool_credential_delete(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.credential import skyvern_credential_delete # noqa: PLC0415
return await skyvern_credential_delete(**kwargs)
@credential_app.callback()
def credential_callback(
api_key: str | None = typer.Option(
@ -31,8 +45,10 @@ def credential_callback(
),
) -> None:
"""Load environment and optional API key override."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
if api_key:
from skyvern.config import settings # noqa: PLC0415
settings.SKYVERN_API_KEY = api_key

View file

@ -13,15 +13,14 @@ from __future__ import annotations
import os
import typer
from dotenv import load_dotenv
from rich.table import Table
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.client import Skyvern
from skyvern.client.types.non_empty_credit_card_credential import NonEmptyCreditCardCredential
from skyvern.client.types.non_empty_password_credential import NonEmptyPasswordCredential
from skyvern.client.types.secret_credential import SecretCredential
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import EnvIntent
from .commands._output import output, output_error
from .commands._tty import is_interactive, require_interactive_or_flag
@ -49,7 +48,9 @@ def credentials_callback(
def _get_client(api_key: str | None = None) -> Skyvern:
"""Instantiate a Skyvern SDK client using environment variables."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
from skyvern.config import settings # noqa: PLC0415
key = api_key or os.getenv("SKYVERN_API_KEY") or settings.SKYVERN_API_KEY
return Skyvern(base_url=settings.SKYVERN_BASE_URL, api_key=key)

View file

@ -1,18 +1,29 @@
import shutil
import subprocess
import time
from typing import Optional
from pathlib import Path
from typing import Any, Optional
from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Confirm
from skyvern.analytics import capture_setup_event
from .console import console
from .llm_setup import DEFAULT_POSTGRES_DATABASE_STRING, update_or_add_env_var
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(event_name, success, error_type, error_message, extra_data)
def command_exists(command: str) -> bool:
return shutil.which(command) is not None
@ -111,7 +122,7 @@ def is_postgres_container_exists() -> bool:
return code == 0
def setup_postgresql(no_postgres: bool = False) -> None:
def setup_postgresql(no_postgres: bool = False, *, env_path: Path | str | None = None) -> None:
"""Set up PostgreSQL database for Skyvern."""
console.print(Panel("[bold cyan]PostgreSQL Setup[/bold cyan]", border_style="blue"))
capture_setup_event("database-start")
@ -123,7 +134,7 @@ def setup_postgresql(no_postgres: bool = False) -> None:
console.print("✅ [green]Database and user exist.[/green]")
else:
create_database_and_user()
update_or_add_env_var("DATABASE_STRING", DEFAULT_POSTGRES_DATABASE_STRING)
update_or_add_env_var("DATABASE_STRING", DEFAULT_POSTGRES_DATABASE_STRING, env_path=env_path)
capture_setup_event("database-complete", success=True, extra_data={"source": "local"})
return
@ -264,5 +275,5 @@ def setup_postgresql(no_postgres: bool = False) -> None:
else:
console.print("✅ [green]Database and user created successfully.[/green]")
update_or_add_env_var("DATABASE_STRING", DEFAULT_POSTGRES_DATABASE_STRING)
update_or_add_env_var("DATABASE_STRING", DEFAULT_POSTGRES_DATABASE_STRING, env_path=env_path)
capture_setup_event("database-complete", success=True, extra_data={"source": "docker"})

View file

@ -4,7 +4,7 @@ import subprocess
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, overload
from typing import Any, Literal, overload
import typer
from rich.padding import Padding
@ -12,8 +12,14 @@ from rich.panel import Panel
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Confirm, Prompt
from skyvern.analytics import capture_setup_event
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import (
BACKEND_ENV_FILE_ENV_VAR,
EnvIntent,
EnvScope,
env_scope_label,
parse_env_scope,
resolve_backend_env_path,
)
from .browser import setup_browser_config
from .console import console
@ -52,6 +58,18 @@ class InitEnvResult:
return self.run_local
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(event_name, success, error_type, error_message, extra_data)
def _browser_type_requires_playwright_chromium(browser_type: str | None) -> bool:
return browser_type in PLAYWRIGHT_CHROMIUM_BROWSER_TYPES
@ -144,11 +162,53 @@ def _init_return_value(result: InitEnvResult, return_result: bool) -> bool | Ini
return result.run_local
def _parse_env_scope(value: str) -> EnvScope:
try:
return parse_env_scope(value)
except ValueError as exc:
raise typer.BadParameter(str(exc)) from exc
def _select_backend_env_scope(*, run_local: bool, env_scope: str | None) -> EnvScope:
if run_local:
if env_scope is None:
return EnvScope.LEGACY
selected_scope = _parse_env_scope(env_scope)
if selected_scope is not EnvScope.LEGACY:
raise typer.BadParameter(
"Self-hosted local server setup writes ./.env. Project/global scopes are for cloud/API config."
)
return selected_scope
if env_scope is not None:
return _parse_env_scope(env_scope)
default_scope = EnvScope.GLOBAL
console.print("\n[bold blue]Backend config location[/bold blue]")
console.print(" [cyan]1.[/cyan] Current directory (./.env)")
console.print(" [cyan]2.[/cyan] Project directory (./.skyvern/.env)")
console.print(" [cyan]3.[/cyan] Global user directory (~/.skyvern/.env)")
while True:
try:
selected = Prompt.ask(
"Where should Skyvern store backend config?",
default="3",
)
except EOFError:
return default_scope
try:
return _parse_env_scope(selected)
except typer.BadParameter as exc:
console.print(f"[red]{exc.message}[/red]")
@overload
def init_env(
no_postgres: bool = False,
database_string: str = "",
skip_browser_install: bool = False,
env_scope: str | None = None,
env_path: Path | str | None = None,
return_result: Literal[False] = False,
) -> bool: ...
@ -158,6 +218,8 @@ def init_env(
no_postgres: bool = False,
database_string: str = "",
skip_browser_install: bool = False,
env_scope: str | None = None,
env_path: Path | str | None = None,
return_result: Literal[True] = True,
) -> InitEnvResult: ...
@ -166,6 +228,8 @@ def init_env(
no_postgres: bool = False,
database_string: str = "",
skip_browser_install: bool = False,
env_scope: str | None = None,
env_path: Path | str | None = None,
return_result: bool = False,
) -> bool | InitEnvResult:
"""Interactive initialization command for Skyvern."""
@ -185,14 +249,29 @@ def init_env(
run_local = infra_choice == "local"
result = InitEnvResult(run_local=run_local)
selected_env_scope = _select_backend_env_scope(run_local=run_local, env_scope=env_scope)
backend_env_path = (
Path(env_path).expanduser()
if env_path is not None
else resolve_backend_env_path(
intent=EnvIntent.SERVER if run_local else EnvIntent.CLOUD,
scope=selected_env_scope,
for_write=True,
)
)
os.environ[BACKEND_ENV_FILE_ENV_VAR] = str(backend_env_path)
console.print(f"[dim]Backend config: {env_scope_label(selected_env_scope)} -> {backend_env_path}[/dim]")
def set_env_var(key: str, value: str) -> None:
update_or_add_env_var(key, value, env_path=backend_env_path)
if run_local:
if database_string:
console.print("🔗 [bold blue]Using custom database connection...[/bold blue]")
update_or_add_env_var("DATABASE_STRING", database_string)
console.print("✅ [green]Database connection string set in .env file.[/green]")
set_env_var("DATABASE_STRING", database_string)
console.print(f"✅ [green]Database connection string set in {backend_env_path}.[/green]")
else:
setup_postgresql(no_postgres)
setup_postgresql(no_postgres, env_path=backend_env_path)
console.print("📊 [bold blue]Running database migrations...[/bold blue]")
from skyvern.utils import migrate_db # noqa: PLC0415
@ -211,7 +290,6 @@ def init_env(
else:
console.print("[red]Failed to generate local organization API key. Please check server logs.[/red]")
backend_env_path = resolve_backend_env_path()
if backend_env_path.exists():
console.print(f"💡 [{backend_env_path}] file already exists.", style="yellow", markup=False)
redo_llm_setup = Confirm.ask(
@ -222,24 +300,24 @@ def init_env(
console.print("[green]Skipping LLM setup.[/green]")
else:
console.print("\n[bold blue]Initializing .env file for LLM providers...[/bold blue]")
setup_llm_providers()
setup_llm_providers(env_path=backend_env_path)
else:
console.print("\n[bold blue]Initializing .env file...[/bold blue]")
setup_llm_providers()
setup_llm_providers(env_path=backend_env_path)
console.print("\n[bold blue]Configuring browser settings...[/bold blue]")
browser_type, browser_location, remote_debugging_url = setup_browser_config()
result.browser_type = browser_type
update_or_add_env_var("BROWSER_TYPE", browser_type)
set_env_var("BROWSER_TYPE", browser_type)
if browser_location:
update_or_add_env_var("CHROME_EXECUTABLE_PATH", browser_location)
set_env_var("CHROME_EXECUTABLE_PATH", browser_location)
if remote_debugging_url:
update_or_add_env_var("BROWSER_REMOTE_DEBUGGING_URL", remote_debugging_url)
update_or_add_env_var("BROWSER_STREAMING_MODE", "cdp")
set_env_var("BROWSER_REMOTE_DEBUGGING_URL", remote_debugging_url)
set_env_var("BROWSER_STREAMING_MODE", "cdp")
console.print("✅ [green]Browser configuration complete.[/green]")
console.print("🌐 [bold blue]Setting Skyvern Base URL to: http://localhost:8000[/bold blue]")
update_or_add_env_var("SKYVERN_BASE_URL", "http://localhost:8000")
set_env_var("SKYVERN_BASE_URL", "http://localhost:8000")
console.print("\n[bold yellow]To run Skyvern you can either:[/bold yellow]")
console.print("• [green]skyvern run server[/green] (reuses the DB we just created)")
@ -267,7 +345,7 @@ def init_env(
default="https://app.skyvern.com",
show_default=True,
)
run_signup(base_url=frontend_url)
run_signup(base_url=frontend_url, env_path=backend_env_path)
api_key = None # already saved by browser_auth
else:
base_url = Prompt.ask("Enter Skyvern base URL", default="https://api.skyvern.com", show_default=True)
@ -285,7 +363,7 @@ def init_env(
if not api_key:
console.print("[bold red]Error: API key cannot be empty. Aborting initialization.[/bold red]")
return _init_return_value(result, return_result)
update_or_add_env_var("SKYVERN_BASE_URL", base_url)
set_env_var("SKYVERN_BASE_URL", base_url)
console.print(
"\n[bold yellow]Tip:[/bold yellow] Want Skyvern Cloud to use your local browser "
@ -297,10 +375,10 @@ def init_env(
analytics_id_input = Prompt.ask("Please enter your email for analytics (press enter to skip)", default="")
analytics_id = analytics_id_input if analytics_id_input else str(uuid.uuid4())
update_or_add_env_var("ANALYTICS_ID", analytics_id)
set_env_var("ANALYTICS_ID", analytics_id)
if api_key:
update_or_add_env_var("SKYVERN_API_KEY", api_key)
console.print(f"✅ [green]{resolve_backend_env_path()} file has been initialized.[/green]")
set_env_var("SKYVERN_API_KEY", api_key)
console.print(f"✅ [green]{backend_env_path} file has been initialized.[/green]")
# Retrieve browser config for MCP setup (set during local init)
_mcp_browser_type = os.environ.get("BROWSER_TYPE") if run_local else None
@ -352,10 +430,15 @@ def init_app_factory() -> typer.Typer:
"--database-string",
help="Custom database connection string (e.g., postgresql+psycopg://user:password@host:port/dbname). When provided, skips Docker PostgreSQL setup.",
),
env_scope: str | None = typer.Option(
None,
"--env-scope",
help="Backend env location: legacy/current, project, or global.",
),
) -> None:
"""Run full initialization when no subcommand is provided."""
if ctx.invoked_subcommand is None:
init_env(no_postgres=no_postgres, database_string=database_string)
init_env(no_postgres=no_postgres, database_string=database_string, env_scope=env_scope)
@app.command(name="browser")
def _init_browser_command() -> None:

View file

@ -1,11 +1,12 @@
import os
from pathlib import Path
from typing import Any
from dotenv import load_dotenv, set_key
from rich.panel import Panel
from rich.prompt import Confirm, Prompt
from skyvern.analytics import capture_setup_event
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import EnvIntent, EnvScope, resolve_backend_env_path
from .console import console
from .masked_prompt import ask_secret
@ -13,11 +14,35 @@ from .masked_prompt import ask_secret
DEFAULT_POSTGRES_DATABASE_STRING = "postgresql+psycopg://skyvern@localhost:5432/skyvern"
def update_or_add_env_var(key: str, value: str) -> None:
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(event_name, success, error_type, error_message, extra_data)
def update_or_add_env_var(
key: str,
value: str,
*,
env_path: Path | str | None = None,
intent: EnvIntent | str = EnvIntent.AUTO,
scope: EnvScope | str | None = None,
) -> None:
"""Update or add environment variable in .env file."""
env_path = resolve_backend_env_path()
if not env_path.exists():
env_path.touch()
resolved_env_path = (
Path(env_path).expanduser()
if env_path is not None
else resolve_backend_env_path(intent=intent, scope=scope, for_write=True)
)
if not resolved_env_path.exists():
resolved_env_path.parent.mkdir(parents=True, exist_ok=True)
resolved_env_path.touch()
defaults = {
"ENV": "local",
"ENABLE_OPENAI": "false",
@ -52,15 +77,19 @@ def update_or_add_env_var(key: str, value: str) -> None:
"ENABLE_LOG_ARTIFACTS": "false",
}
for k, v in defaults.items():
set_key(env_path, k, v)
set_key(resolved_env_path, k, v)
load_dotenv(env_path)
set_key(env_path, key, value)
load_dotenv(resolved_env_path)
set_key(resolved_env_path, key, value)
os.environ[key] = value
def setup_llm_providers() -> None:
def setup_llm_providers(env_path: Path | str | None = None) -> None:
"""Configure Large Language Model (LLM) Providers."""
def set_env_var(key: str, value: str) -> None:
update_or_add_env_var(key, value, env_path=env_path)
console.print(Panel("[bold magenta]LLM Provider Configuration[/bold magenta]", border_style="purple"))
console.print("[italic]Note: All information provided here will be stored only on your local machine.[/italic]")
capture_setup_event("llm-start")
@ -75,8 +104,8 @@ def setup_llm_providers() -> None:
if not openai_api_key:
console.print("[red]Error: OpenAI API key is required. OpenAI will not be enabled.[/red]")
else:
update_or_add_env_var("OPENAI_API_KEY", openai_api_key)
update_or_add_env_var("ENABLE_OPENAI", "true")
set_env_var("OPENAI_API_KEY", openai_api_key)
set_env_var("ENABLE_OPENAI", "true")
enabled_providers.append("openai")
model_options.extend(
[
@ -86,7 +115,7 @@ def setup_llm_providers() -> None:
]
)
else:
update_or_add_env_var("ENABLE_OPENAI", "false")
set_env_var("ENABLE_OPENAI", "false")
console.print("\n[bold blue]--- Anthropic Configuration ---[/bold blue]")
console.print("To enable Anthropic, you must have an Anthropic API key.")
@ -96,8 +125,8 @@ def setup_llm_providers() -> None:
if not anthropic_api_key:
console.print("[red]Error: Anthropic API key is required. Anthropic will not be enabled.[/red]")
else:
update_or_add_env_var("ANTHROPIC_API_KEY", anthropic_api_key)
update_or_add_env_var("ENABLE_ANTHROPIC", "true")
set_env_var("ANTHROPIC_API_KEY", anthropic_api_key)
set_env_var("ENABLE_ANTHROPIC", "true")
enabled_providers.append("anthropic")
model_options.extend(
[
@ -109,7 +138,7 @@ def setup_llm_providers() -> None:
]
)
else:
update_or_add_env_var("ENABLE_ANTHROPIC", "false")
set_env_var("ENABLE_ANTHROPIC", "false")
console.print("\n[bold blue]--- Azure Configuration ---[/bold blue]")
console.print("To enable Azure, you must have an Azure deployment name, API key, base URL, and API version.")
@ -122,15 +151,15 @@ def setup_llm_providers() -> None:
if not all([azure_deployment, azure_api_key, azure_api_base, azure_api_version]):
console.print("[red]Error: All Azure fields must be populated. Azure will not be enabled.[/red]")
else:
update_or_add_env_var("AZURE_DEPLOYMENT", azure_deployment)
update_or_add_env_var("AZURE_API_KEY", azure_api_key)
update_or_add_env_var("AZURE_API_BASE", azure_api_base)
update_or_add_env_var("AZURE_API_VERSION", azure_api_version)
update_or_add_env_var("ENABLE_AZURE", "true")
set_env_var("AZURE_DEPLOYMENT", azure_deployment)
set_env_var("AZURE_API_KEY", azure_api_key)
set_env_var("AZURE_API_BASE", azure_api_base)
set_env_var("AZURE_API_VERSION", azure_api_version)
set_env_var("ENABLE_AZURE", "true")
enabled_providers.append("azure")
model_options.append("AZURE_OPENAI")
else:
update_or_add_env_var("ENABLE_AZURE", "false")
set_env_var("ENABLE_AZURE", "false")
console.print("\n[bold blue]--- Gemini Configuration ---[/bold blue]")
console.print("To enable Gemini, you must have a Gemini API key.")
@ -140,8 +169,8 @@ def setup_llm_providers() -> None:
if not gemini_api_key:
console.print("[red]Error: Gemini API key is required. Gemini will not be enabled.[/red]")
else:
update_or_add_env_var("GEMINI_API_KEY", gemini_api_key)
update_or_add_env_var("ENABLE_GEMINI", "true")
set_env_var("GEMINI_API_KEY", gemini_api_key)
set_env_var("ENABLE_GEMINI", "true")
enabled_providers.append("gemini")
model_options.extend(
[
@ -154,7 +183,7 @@ def setup_llm_providers() -> None:
]
)
else:
update_or_add_env_var("ENABLE_GEMINI", "false")
set_env_var("ENABLE_GEMINI", "false")
console.print("\n[bold blue]--- Ollama / Local LLM Configuration ---[/bold blue]")
console.print("Use any locally-running model via Ollama (e.g. gemma4, qwen3, deepseek-r1).")
@ -172,14 +201,14 @@ def setup_llm_providers() -> None:
console.print("[red]Error: Model name is required. Ollama will not be enabled.[/red]")
else:
ollama_vision = Confirm.ask("Does this model support vision?", default=False)
update_or_add_env_var("OLLAMA_SERVER_URL", ollama_server_url)
update_or_add_env_var("OLLAMA_MODEL", ollama_model)
update_or_add_env_var("OLLAMA_SUPPORTS_VISION", str(ollama_vision).lower())
update_or_add_env_var("ENABLE_OLLAMA", "true")
set_env_var("OLLAMA_SERVER_URL", ollama_server_url)
set_env_var("OLLAMA_MODEL", ollama_model)
set_env_var("OLLAMA_SUPPORTS_VISION", str(ollama_vision).lower())
set_env_var("ENABLE_OLLAMA", "true")
enabled_providers.append("ollama")
model_options.append("OLLAMA")
else:
update_or_add_env_var("ENABLE_OLLAMA", "false")
set_env_var("ENABLE_OLLAMA", "false")
if not model_options:
capture_setup_event(
@ -208,7 +237,7 @@ def setup_llm_providers() -> None:
)
chosen_model = model_options[int(chosen_model_idx) - 1]
console.print(f"🎉 [bold green]Chosen LLM Model: {chosen_model}[/bold green]")
update_or_add_env_var("LLM_KEY", chosen_model)
set_env_var("LLM_KEY", chosen_model)
capture_setup_event(
"llm-complete",
success=True,

View file

@ -5,6 +5,7 @@ import subprocess
import sys
from enum import Enum
from pathlib import Path
from typing import Any
import typer
from rich.panel import Panel
@ -12,13 +13,35 @@ from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.prompt import Confirm, Prompt
from rich.text import Text
from skyvern.analytics import capture_setup_error, capture_setup_event
from skyvern.cli.console import console
from skyvern.utils.env_paths import resolve_frontend_env_path
from skyvern.utils.env_paths import EnvScope, parse_env_scope, resolve_frontend_env_path
quickstart_app = typer.Typer(help="Quickstart command to set up and run Skyvern with one command.")
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(event_name, success, error_type, error_message, extra_data)
def capture_setup_error(
event_name: str,
error: Exception,
error_type: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
from skyvern.analytics import capture_setup_error as _capture_setup_error # noqa: PLC0415
_capture_setup_error(event_name, error, error_type, extra_data)
class QuickstartPath(str, Enum):
CLOUD = "cloud"
LOCAL = "local"
@ -155,6 +178,13 @@ def _server_quickstart_flags_requested(
return no_postgres or bool(database_string) or skip_browser_install or server_only
def _parse_quickstart_env_scope(value: str) -> EnvScope:
try:
return parse_env_scope(value)
except ValueError as exc:
raise typer.BadParameter(str(exc)) from exc
def _print_cloud_guidance() -> None:
message = """Cloud/API SDK usage
@ -479,11 +509,17 @@ def quickstart(
callback=_validate_install_type,
help="Choose quickstart path: cloud, local, or server.",
),
env_scope: str | None = typer.Option(
None,
"--env-scope",
help="Backend env location for setup writes: legacy/current, project, or global.",
),
) -> None:
"""Quickstart command to set up and run Skyvern with one command."""
# Run initialization
console.print(Panel("[bold green]🚀 Starting Skyvern Quickstart[/bold green]", border_style="green"))
install_type_value = install_type if isinstance(install_type, str) else None
env_scope_value = env_scope if isinstance(env_scope, str) else None
has_server_extra = _has_server_quickstart_extra()
# The server extra is a superset of the local embedded runtime dependencies.
@ -508,6 +544,16 @@ def quickstart(
)
)
raise typer.Exit(1)
if env_scope_value is not None and _parse_quickstart_env_scope(env_scope_value) is not EnvScope.LEGACY:
console.print(
Panel(
"[bold red]Conflicting quickstart options.[/bold red]\n"
"Docker Compose uses the source checkout `.env` file. "
"Use `--env-scope legacy` or omit `--env-scope`.",
border_style="red",
)
)
raise typer.Exit(1)
elif install_type_value is None and server_flags_requested:
selected_path = QuickstartPath.SERVER
else:
@ -522,6 +568,16 @@ def quickstart(
if selected_path is QuickstartPath.LOCAL:
_print_local_guidance(has_local_extra=has_local_extra)
raise typer.Exit(0)
if env_scope_value is not None and _parse_quickstart_env_scope(env_scope_value) is not EnvScope.LEGACY:
console.print(
Panel(
"[bold red]Conflicting quickstart options.[/bold red]\n"
"Self-hosted local server setup writes ./.env. "
"Project/global scopes are for cloud/API config.",
border_style="red",
)
)
raise typer.Exit(1)
# Check if Docker Compose option was explicitly requested or offer choice
docker_compose_available = check_docker_compose_file()

View file

@ -14,21 +14,24 @@ if TYPE_CHECKING:
import psutil
import typer
import uvicorn
from dotenv import load_dotenv, set_key
from dotenv import set_key
from rich.panel import Panel
from rich.prompt import Confirm
from starlette.middleware import Middleware
from starlette.responses import JSONResponse as StarletteJSONResponse
from starlette.responses import Response as StarletteResponse
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.cli.commands._output import output_error
from skyvern.cli.commands._tty import is_interactive
from skyvern.cli.console import console
from skyvern.cli.core.result import set_concise_responses
from skyvern.cli.utils import start_services
from skyvern.config import settings
from skyvern.utils import detect_os
from skyvern.utils.env_paths import resolve_backend_env_path, resolve_frontend_env_path
from skyvern.utils.env_paths import (
EnvIntent,
resolve_backend_env_path,
resolve_frontend_env_path,
)
run_app = typer.Typer(help="Commands to run Skyvern services such as the API server or UI.")
_mcp_cleanup_done = False
@ -116,7 +119,7 @@ def run_server() -> None:
_handle_missing_dep(exc)
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.SERVER)
from skyvern.config import settings # noqa: PLC0415
port = settings.PORT
@ -189,9 +192,9 @@ def run_ui(
shutil.copy(frontend_dir / ".env.example", frontend_env_path)
console.print("✅ [green]Successfully set up frontend .env file[/green]")
backend_env_path = resolve_backend_env_path()
backend_env_path = resolve_backend_env_path(intent=EnvIntent.SERVER)
if backend_env_path.exists():
load_dotenv(backend_env_path)
prepare_cli_runtime(intent=EnvIntent.SERVER)
skyvern_api_key = os.getenv("SKYVERN_API_KEY")
if skyvern_api_key:
set_key(frontend_env_path, "VITE_SKYVERN_API_KEY", skyvern_api_key)
@ -326,6 +329,8 @@ def run_docker() -> None:
@run_app.command(name="all")
def run_all() -> None:
"""Run the Skyvern API server and UI server in parallel."""
from skyvern.cli.utils import start_services # noqa: PLC0415
asyncio.run(start_services())
@ -336,7 +341,7 @@ def run_dev() -> None:
This command starts both services and immediately returns control to your terminal.
Use 'skyvern stop all' to stop the services.
"""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.SERVER)
from skyvern.config import settings as skyvern_settings # noqa: PLC0415
console.print(Panel("[bold green]Starting Skyvern in development mode...[/bold green]", border_style="green"))
@ -442,6 +447,7 @@ def run_mcp(
] = False,
) -> None:
"""Run the MCP server with configurable transport for local or remote hosting."""
prepare_cli_runtime(intent=EnvIntent.CLOUD)
from skyvern.cli.core.mcp_http_auth import MCPAPIKeyMiddleware # noqa: PLC0415
from skyvern.cli.core.session_manager import set_stateless_http_mode # noqa: PLC0415
from skyvern.cli.mcp_tools import mcp # noqa: PLC0415
@ -531,6 +537,9 @@ def run_code(
logging.getLogger("LiteLLM").setLevel(logging.CRITICAL)
logging.getLogger("LiteLLM Router").setLevel(logging.CRITICAL)
logging.getLogger("LiteLLM Proxy").setLevel(logging.CRITICAL)
from skyvern.config import settings # noqa: PLC0415
settings.LOG_LEVEL = "CRITICAL"
from skyvern.forge.sdk.forge_log import setup_logger # noqa: PLC0415

View file

@ -7,20 +7,11 @@ from typing import Any
import typer
from click.core import ParameterSource
from dotenv import load_dotenv
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.utils.env_paths import EnvIntent
from .commands._output import run_tool
from .mcp_tools.schedule import skyvern_schedule_create as tool_schedule_create
from .mcp_tools.schedule import skyvern_schedule_delete as tool_schedule_delete
from .mcp_tools.schedule import skyvern_schedule_disable as tool_schedule_disable
from .mcp_tools.schedule import skyvern_schedule_enable as tool_schedule_enable
from .mcp_tools.schedule import skyvern_schedule_get as tool_schedule_get
from .mcp_tools.schedule import skyvern_schedule_list as tool_schedule_list
from .mcp_tools.schedule import skyvern_schedule_list_for_workflow as tool_schedule_list_for_workflow
from .mcp_tools.schedule import skyvern_schedule_update as tool_schedule_update
schedule_app = typer.Typer(
help="Manage workflow schedules (list, create, update, enable/disable, delete).",
@ -28,6 +19,54 @@ schedule_app = typer.Typer(
)
async def tool_schedule_list(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_list # noqa: PLC0415
return await skyvern_schedule_list(**kwargs)
async def tool_schedule_list_for_workflow(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_list_for_workflow # noqa: PLC0415
return await skyvern_schedule_list_for_workflow(**kwargs)
async def tool_schedule_get(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_get # noqa: PLC0415
return await skyvern_schedule_get(**kwargs)
async def tool_schedule_create(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_create # noqa: PLC0415
return await skyvern_schedule_create(**kwargs)
async def tool_schedule_update(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_update # noqa: PLC0415
return await skyvern_schedule_update(**kwargs)
async def tool_schedule_enable(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_enable # noqa: PLC0415
return await skyvern_schedule_enable(**kwargs)
async def tool_schedule_disable(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_disable # noqa: PLC0415
return await skyvern_schedule_disable(**kwargs)
async def tool_schedule_delete(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.schedule import skyvern_schedule_delete # noqa: PLC0415
return await skyvern_schedule_delete(**kwargs)
@schedule_app.callback()
def schedule_callback(
api_key: str | None = typer.Option(
@ -37,8 +76,10 @@ def schedule_callback(
help="Skyvern API key.",
),
) -> None:
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
if api_key:
from skyvern.config import settings # noqa: PLC0415
settings.SKYVERN_API_KEY = api_key
@ -104,6 +145,7 @@ def schedule_list(
async def _run() -> dict[str, Any]:
if workflow_id is not None:
return await tool_schedule_list_for_workflow(workflow_permanent_id=workflow_id)
return await tool_schedule_list(page=page, page_size=page_size, status=status, search=search)
run_tool(

View file

@ -12,23 +12,21 @@ import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, cast
from typing import Any, Callable, cast
from urllib.parse import urlparse
import json5
import typer
import yaml
from dotenv import load_dotenv
from rich.panel import Panel
from rich.syntax import Syntax
from rich.table import Table
from skyvern.analytics import capture_setup_event
from skyvern.cli.auth_command import run_signup
from skyvern.cli.console import console
from skyvern.cli.skill_commands import get_skill_dirs
from skyvern.utils import detect_os, get_windows_appdata_roaming
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import EnvIntent, load_backend_env_files
# NOTE: These helpers back both `skyvern setup ...` commands and the
# interactive MCP step used by `skyvern init` / `skyvern quickstart`, plus
@ -56,10 +54,8 @@ _JSON5_SINGLE_QUOTED_STRING_RE = re.compile(r"(^|[:[{,]\s*)'(?:[^'\\]|\\.)*'", r
def _get_env_credentials() -> tuple[str, str]:
"""Read SKYVERN_API_KEY and SKYVERN_BASE_URL from environment or .env."""
backend_env = resolve_backend_env_path()
if backend_env.exists():
load_dotenv(backend_env, override=False)
"""Read cloud SKYVERN_API_KEY and SKYVERN_BASE_URL from environment or scoped env files."""
load_backend_env_files(intent=EnvIntent.CLOUD)
api_key = os.environ.get("SKYVERN_API_KEY", "")
base_url = os.environ.get("SKYVERN_BASE_URL", "https://api.skyvern.com")
@ -67,16 +63,33 @@ def _get_env_credentials() -> tuple[str, str]:
def _get_local_env_credentials() -> tuple[str, str]:
"""Read local SKYVERN_API_KEY and SKYVERN_BASE_URL from environment or .env."""
backend_env = resolve_backend_env_path()
if backend_env.exists():
load_dotenv(backend_env, override=False)
"""Read local SKYVERN_API_KEY and SKYVERN_BASE_URL from legacy server env."""
load_backend_env_files(intent=EnvIntent.SERVER)
api_key = os.environ.get("SKYVERN_API_KEY", "")
base_url = os.environ.get("SKYVERN_BASE_URL", "")
return api_key, base_url
def capture_setup_event(
event_name: str,
success: bool = True,
error_type: str | None = None,
error_message: str | None = None,
extra_data: dict[str, Any] | None = None,
) -> None:
"""Capture setup analytics only after setup env intent has been selected."""
from skyvern.analytics import capture_setup_event as _capture_setup_event # noqa: PLC0415
_capture_setup_event(
event_name=event_name,
success=success,
error_type=error_type,
error_message=error_message,
extra_data=extra_data,
)
def _build_remote_mcp_entry(api_key: str, url: str = _DEFAULT_REMOTE_URL) -> dict:
"""Build an HTTP MCP entry for remote/cloud hosting."""
entry: dict = {
@ -1265,13 +1278,10 @@ def setup_hermes(
url: str | None = _url_opt,
) -> None:
"""Register Skyvern MCP with Hermes (remote by default, --local for stdio)."""
env_key, env_base_url = _get_env_credentials()
resolved_key = api_key or env_key
if local:
# Local stdio mode: Hermes spawns `skyvern run mcp` as a child process
local_key, local_base_url = _get_local_env_credentials()
resolved_local_key = api_key or local_key or resolved_key or ""
resolved_local_key = api_key or local_key or ""
resolved_base_url = local_base_url or ""
if not resolved_base_url:
console.print(
@ -1292,8 +1302,11 @@ def setup_hermes(
}
if local_entry.get("env"):
hermes_entry["env"] = local_entry["env"]
resolved_key = resolved_local_key
else:
# Remote HTTP mode: Hermes connects to hosted MCP server
env_key, _ = _get_env_credentials()
resolved_key = api_key or env_key
if not resolved_key:
console.print(
"[red]No API key found. Run [bold]skyvern login[/bold] or set "

View file

@ -6,12 +6,11 @@ import json
import os
import typer
from dotenv import load_dotenv
from rich.panel import Panel
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.client import Skyvern
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import EnvIntent
from .commands._output import output, output_error
from .console import console
@ -36,7 +35,9 @@ def tasks_callback(
def _get_client(api_key: str | None = None) -> Skyvern:
"""Instantiate a Skyvern SDK client using environment variables."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
from skyvern.config import settings # noqa: PLC0415
key = api_key or os.getenv("SKYVERN_API_KEY") or settings.SKYVERN_API_KEY
return Skyvern(base_url=settings.SKYVERN_BASE_URL, api_key=key)

View file

@ -5,24 +5,63 @@ from __future__ import annotations
from typing import Any
import typer
from dotenv import load_dotenv
from skyvern.config import settings
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern._cli_bootstrap import prepare_cli_runtime
from skyvern.utils.env_paths import EnvIntent
from .commands._output import resolve_inline_or_file, run_tool
from .mcp_tools.workflow import skyvern_workflow_cancel as tool_workflow_cancel
from .mcp_tools.workflow import skyvern_workflow_create as tool_workflow_create
from .mcp_tools.workflow import skyvern_workflow_delete as tool_workflow_delete
from .mcp_tools.workflow import skyvern_workflow_get as tool_workflow_get
from .mcp_tools.workflow import skyvern_workflow_list as tool_workflow_list
from .mcp_tools.workflow import skyvern_workflow_run as tool_workflow_run
from .mcp_tools.workflow import skyvern_workflow_status as tool_workflow_status
from .mcp_tools.workflow import skyvern_workflow_update as tool_workflow_update
workflow_app = typer.Typer(help="Manage Skyvern workflows.", no_args_is_help=True)
async def tool_workflow_list(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_list # noqa: PLC0415
return await skyvern_workflow_list(**kwargs)
async def tool_workflow_get(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_get # noqa: PLC0415
return await skyvern_workflow_get(**kwargs)
async def tool_workflow_create(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_create # noqa: PLC0415
return await skyvern_workflow_create(**kwargs)
async def tool_workflow_update(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_update # noqa: PLC0415
return await skyvern_workflow_update(**kwargs)
async def tool_workflow_delete(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_delete # noqa: PLC0415
return await skyvern_workflow_delete(**kwargs)
async def tool_workflow_run(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_run # noqa: PLC0415
return await skyvern_workflow_run(**kwargs)
async def tool_workflow_status(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_status # noqa: PLC0415
return await skyvern_workflow_status(**kwargs)
async def tool_workflow_cancel(**kwargs: Any) -> dict[str, Any]:
from .mcp_tools.workflow import skyvern_workflow_cancel # noqa: PLC0415
return await skyvern_workflow_cancel(**kwargs)
@workflow_app.callback()
def workflow_callback(
api_key: str | None = typer.Option(
@ -33,8 +72,10 @@ def workflow_callback(
),
) -> None:
"""Load workflow CLI environment and optional API key override."""
load_dotenv(resolve_backend_env_path())
prepare_cli_runtime(intent=EnvIntent.CLOUD)
if api_key:
from skyvern.config import settings # noqa: PLC0415
settings.SKYVERN_API_KEY = api_key

View file

@ -1,4 +1,5 @@
import logging
import os
import platform
from pathlib import Path
from typing import Any
@ -8,7 +9,12 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
from skyvern import constants
from skyvern.constants import REPO_ROOT_DIR, SKYVERN_DIR
from skyvern.utils.env_paths import resolve_backend_env_path
from skyvern.utils.env_paths import (
BACKEND_ENV_BASENAMES,
BACKEND_ENV_INTENT_ENV_VAR,
EnvIntent,
backend_env_path_candidates,
)
def _default_database_string() -> str:
@ -43,10 +49,23 @@ def _ensure_sqlite_dir(database_string: str) -> None:
# Even if we were to resolve paths at instantiation time, the global `settings`
# singleton instantiation at the bottom of this file also runs at import time
# and relies on the same assumption.
_DEFAULT_ENV_FILES = (
resolve_backend_env_path(".env"),
resolve_backend_env_path(".env.staging"),
resolve_backend_env_path(".env.prod"),
#
# pydantic-settings applies later dotenv files with higher precedence, so the
# resolver's read-priority order is reversed here. With no explicit CLI intent,
# AUTO preserves legacy self-hosted imports by only considering ./.env.
def _settings_env_intent() -> EnvIntent:
try:
return EnvIntent(os.getenv(BACKEND_ENV_INTENT_ENV_VAR, EnvIntent.AUTO.value))
except ValueError:
return EnvIntent.AUTO
def _settings_env_file_candidates(basename: str) -> tuple[Path, ...]:
return tuple(reversed(backend_env_path_candidates(basename, intent=_settings_env_intent())))
_DEFAULT_ENV_FILES = tuple(
candidate for basename in BACKEND_ENV_BASENAMES for candidate in _settings_env_file_candidates(basename)
)

View file

@ -1,16 +1,212 @@
import os
from enum import Enum
from pathlib import Path
from typing import Optional
BACKEND_ENV_DEFAULT = ".env"
BACKEND_ENV_BASENAMES = (BACKEND_ENV_DEFAULT, ".env.staging", ".env.prod")
BACKEND_ENV_DIRNAME = ".skyvern"
BACKEND_ENV_FILE_ENV_VAR = "SKYVERN_ENV_FILE"
BACKEND_ENV_INTENT_ENV_VAR = "SKYVERN_ENV_INTENT"
FRONTEND_DIRNAME = "skyvern-frontend"
FRONTEND_ENV_FILENAME = ".env"
class EnvScope(str, Enum):
LEGACY = "legacy"
PROJECT = "project"
GLOBAL = "global"
class EnvIntent(str, Enum):
AUTO = "auto"
CLOUD = "cloud"
LOCAL = "local"
SERVER = "server"
_READ_SCOPE_ORDER: dict[EnvIntent, tuple[EnvScope, ...]] = {
# Unscoped imports preserve legacy self-hosted behavior until a CLI command
# selects a cloud/local intent explicitly.
EnvIntent.AUTO: (EnvScope.LEGACY,),
EnvIntent.CLOUD: (EnvScope.PROJECT, EnvScope.GLOBAL, EnvScope.LEGACY),
# Reserved for embedded local SDK config writers; self-hosted CLI setup uses SERVER.
EnvIntent.LOCAL: (EnvScope.PROJECT, EnvScope.LEGACY, EnvScope.GLOBAL),
EnvIntent.SERVER: (EnvScope.LEGACY,),
}
_WRITE_SCOPE_DEFAULTS: dict[EnvIntent, EnvScope] = {
EnvIntent.AUTO: EnvScope.LEGACY,
EnvIntent.CLOUD: EnvScope.GLOBAL,
EnvIntent.LOCAL: EnvScope.PROJECT,
EnvIntent.SERVER: EnvScope.LEGACY,
}
_ENV_SCOPE_ALIASES = {
"1": EnvScope.LEGACY,
"legacy": EnvScope.LEGACY,
"cwd": EnvScope.LEGACY,
"current": EnvScope.LEGACY,
"2": EnvScope.PROJECT,
"project": EnvScope.PROJECT,
"3": EnvScope.GLOBAL,
"global": EnvScope.GLOBAL,
"user": EnvScope.GLOBAL,
}
def _normalize_env_scope(scope: EnvScope | str | None) -> EnvScope | None:
if scope is None:
return None
if isinstance(scope, EnvScope):
return scope
normalized = scope.strip().lower()
choice = _ENV_SCOPE_ALIASES.get(normalized)
if choice is not None:
return choice
raise ValueError("Choose one of: legacy/current, project, global, 1, 2, or 3.")
def _normalize_env_intent(intent: EnvIntent | str) -> EnvIntent:
if isinstance(intent, EnvIntent):
return intent
return EnvIntent(intent.strip().lower())
def parse_env_scope(value: str) -> EnvScope:
choice = _ENV_SCOPE_ALIASES.get(value.strip().lower())
if choice is None:
raise ValueError("Choose one of: legacy/current, project, global, 1, 2, or 3.")
return choice
def _explicit_backend_env_path(basename: str) -> Path | None:
explicit_path = os.getenv(BACKEND_ENV_FILE_ENV_VAR)
if not explicit_path:
return None
env_path = Path(explicit_path).expanduser()
if basename == BACKEND_ENV_DEFAULT:
return env_path
return env_path.parent / basename
def backend_env_path_for_scope(scope: EnvScope | str, basename: str = BACKEND_ENV_DEFAULT) -> Path:
"""Return the backend env path for an explicit storage scope."""
normalized_scope = _normalize_env_scope(scope)
if normalized_scope is EnvScope.LEGACY:
return Path.cwd() / basename
if normalized_scope is EnvScope.PROJECT:
return Path.cwd() / BACKEND_ENV_DIRNAME / basename
if normalized_scope is EnvScope.GLOBAL:
return Path.home() / BACKEND_ENV_DIRNAME / basename
raise ValueError(f"Unsupported env scope: {scope}")
def backend_env_path_candidates(
basename: str = BACKEND_ENV_DEFAULT,
*,
intent: EnvIntent | str = EnvIntent.AUTO,
) -> tuple[Path, ...]:
explicit_path = _explicit_backend_env_path(basename)
if explicit_path is not None:
return (explicit_path,)
normalized_intent = _normalize_env_intent(intent)
return tuple(backend_env_path_for_scope(scope, basename) for scope in _READ_SCOPE_ORDER[normalized_intent])
def _backend_env_load_candidates(intent: EnvIntent) -> tuple[Path, ...]:
explicit_path = _explicit_backend_env_path(BACKEND_ENV_DEFAULT)
if explicit_path is not None:
explicit_paths: list[Path] = []
for basename in BACKEND_ENV_BASENAMES:
basename_path = _explicit_backend_env_path(basename)
if basename_path is not None:
explicit_paths.append(basename_path)
return tuple(explicit_paths)
paths: list[Path] = []
for scope in reversed(_READ_SCOPE_ORDER[intent]):
paths.extend(backend_env_path_for_scope(scope, basename) for basename in BACKEND_ENV_BASENAMES)
return tuple(paths)
def load_backend_env_files(
*,
intent: EnvIntent | str = EnvIntent.AUTO,
override: bool = False,
) -> Path:
"""Load backend env files for an intent and return the highest-priority path.
Files are layered from lowest to highest priority so a project env can
override global defaults, while real process env vars still win by default.
This also records the selected intent in ``SKYVERN_ENV_INTENT`` so later
settings imports use the same precedence.
This mutates process-global environment state. CLI entrypoints should call
it once, before importing ``skyvern.config`` or modules that import it, and
avoid mixing cloud/server intent helpers in the same command.
"""
from dotenv import dotenv_values # noqa: PLC0415
normalized_intent = _normalize_env_intent(intent)
os.environ[BACKEND_ENV_INTENT_ENV_VAR] = normalized_intent.value
values: dict[str, str] = {}
for candidate in _backend_env_load_candidates(normalized_intent):
if candidate.exists():
values.update({key: value for key, value in dotenv_values(candidate).items() if value is not None})
for key, value in values.items():
if override or key not in os.environ:
os.environ[key] = value
return resolve_backend_env_path(intent=normalized_intent)
def resolve_backend_env_path(
basename: str = BACKEND_ENV_DEFAULT,
*,
intent: EnvIntent | str = EnvIntent.AUTO,
scope: EnvScope | str | None = None,
for_write: bool = False,
) -> Path:
"""Return the backend env file path inside the current working directory."""
return Path.cwd() / basename
"""Resolve the backend env file path.
Resolution keeps source-checkout and existing CLI behavior compatible by
preferring an existing ``./.env`` for reads. New flows can choose a scope
explicitly, and ``SKYVERN_ENV_FILE`` remains the strongest override.
"""
explicit_path = _explicit_backend_env_path(basename)
if explicit_path is not None:
return explicit_path
normalized_scope = _normalize_env_scope(scope)
if normalized_scope is not None:
return backend_env_path_for_scope(normalized_scope, basename)
normalized_intent = _normalize_env_intent(intent)
if for_write:
return backend_env_path_for_scope(_WRITE_SCOPE_DEFAULTS[normalized_intent], basename)
for candidate in backend_env_path_candidates(basename, intent=normalized_intent):
if candidate.exists():
return candidate
fallback_scope = _WRITE_SCOPE_DEFAULTS[normalized_intent]
return backend_env_path_for_scope(fallback_scope, basename)
def env_scope_label(scope: EnvScope | str) -> str:
normalized_scope = _normalize_env_scope(scope)
if normalized_scope is EnvScope.LEGACY:
return "Current directory (./.env)"
if normalized_scope is EnvScope.PROJECT:
return "Project directory (./.skyvern/.env)"
if normalized_scope is EnvScope.GLOBAL:
return "Global user directory (~/.skyvern/.env)"
raise ValueError(f"Unsupported env scope: {scope}")
def resolve_frontend_env_path() -> Optional[Path]:

View file

@ -5,7 +5,9 @@ from __future__ import annotations
import http.server
import threading
import urllib.parse
from pathlib import Path
from skyvern.cli import auth_command
from skyvern.cli.auth_command import _CallbackHandler, _derive_api_base_url, _find_free_port
@ -42,6 +44,57 @@ class TestFindFreePort:
sock.close()
def test_run_signup_defaults_to_cloud_write_path(tmp_path, monkeypatch) -> None:
class FakeSocket:
def getsockname(self) -> tuple[str, int]:
return ("127.0.0.1", 43210)
class FakeServer:
def __init__(self, *_args, **_kwargs) -> None:
self.auth_result = {"api_key": None, "organization_id": None, "email": None}
def server_activate(self) -> None:
pass
def serve_forever(self) -> None:
pass
def shutdown(self) -> None:
pass
class FakeEvent:
def wait(self, timeout: int | None = None) -> bool:
return True
class FakeThread:
def __init__(self, target, daemon: bool) -> None:
self.server = target.__self__
def start(self) -> None:
self.server.auth_result = {"api_key": "sk_test", "organization_id": None, "email": None}
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("HOME", str(tmp_path / "home"))
monkeypatch.delenv("SKYVERN_ENV_FILE", raising=False)
(tmp_path / ".env").write_text("SKYVERN_API_KEY=server-key\n")
saved: list[tuple[str, str, Path]] = []
def fake_update_or_add_env_var(key: str, value: str, env_path: Path) -> None:
saved.append((key, value, Path(env_path)))
monkeypatch.setattr(auth_command, "_find_free_port", lambda: FakeSocket())
monkeypatch.setattr(auth_command.http.server, "HTTPServer", FakeServer)
monkeypatch.setattr(auth_command.threading, "Event", FakeEvent)
monkeypatch.setattr(auth_command.threading, "Thread", FakeThread)
monkeypatch.setattr(auth_command.webbrowser, "open", lambda _url: True)
monkeypatch.setattr("skyvern.cli.llm_setup.update_or_add_env_var", fake_update_or_add_env_var)
assert auth_command.run_signup(timeout=1) == "sk_test"
assert {item[2] for item in saved} == {tmp_path / "home" / ".skyvern" / ".env"}
class TestCallbackHandlerStateValidation:
def _make_server(self, state: str) -> http.server.HTTPServer:
sock = _find_free_port()

View file

@ -1,24 +1,23 @@
from __future__ import annotations
import json
import logging
from types import SimpleNamespace
import subprocess
import sys
import types
import skyvern._cli_bootstrap as cli_bootstrap
from skyvern.utils.env_paths import BACKEND_ENV_INTENT_ENV_VAR, EnvIntent
def test_bootstrap_defaults_to_warning_without_explicit_log_level(monkeypatch) -> None:
setup_calls: list[str] = []
fake_settings = SimpleNamespace(LOG_LEVEL="INFO", model_fields_set=set())
monkeypatch.setattr("skyvern.config.settings", fake_settings)
monkeypatch.setattr("skyvern.forge.sdk.forge_log.setup_logger", lambda: setup_calls.append("called"))
monkeypatch.delenv("LOG_LEVEL", raising=False)
logger_names = ("", "skyvern", "httpx", "litellm", "playwright", "httpcore")
previous_levels = {name: logging.getLogger(name).level for name in logger_names}
try:
cli_bootstrap.configure_cli_bootstrap_logging()
assert setup_calls == ["called"]
assert fake_settings.LOG_LEVEL == "WARNING"
assert "LOG_LEVEL" not in cli_bootstrap.os.environ
for name in logger_names:
assert logging.getLogger(name).level == logging.WARNING
finally:
@ -27,20 +26,74 @@ def test_bootstrap_defaults_to_warning_without_explicit_log_level(monkeypatch) -
def test_bootstrap_honors_explicit_log_level(monkeypatch) -> None:
setup_calls: list[str] = []
fake_settings = SimpleNamespace(LOG_LEVEL="DEBUG", model_fields_set={"LOG_LEVEL"})
monkeypatch.setattr("skyvern.config.settings", fake_settings)
monkeypatch.setattr("skyvern.forge.sdk.forge_log.setup_logger", lambda: setup_calls.append("called"))
monkeypatch.setenv("LOG_LEVEL", "DEBUG")
logger_names = ("", "skyvern", "httpx", "litellm", "playwright", "httpcore")
previous_levels = {name: logging.getLogger(name).level for name in logger_names}
try:
cli_bootstrap.configure_cli_bootstrap_logging()
assert setup_calls == ["called"]
assert fake_settings.LOG_LEVEL == "DEBUG"
assert cli_bootstrap.os.environ["LOG_LEVEL"] == "DEBUG"
for name in logger_names:
assert logging.getLogger(name).level == logging.DEBUG
finally:
for name, level in previous_levels.items():
logging.getLogger(name).setLevel(level)
def test_bootstrap_logging_does_not_import_settings() -> None:
script = """
import json
import sys
from skyvern._cli_bootstrap import configure_cli_bootstrap_logging
before = "skyvern.config" in sys.modules
configure_cli_bootstrap_logging()
after = "skyvern.config" in sys.modules
print(json.dumps({"before": before, "after": after}))
"""
result = subprocess.run([sys.executable, "-c", script], text=True, capture_output=True, check=True)
assert json.loads(result.stdout) == {"before": False, "after": False}
def test_runtime_logging_lazily_calls_setup_logger(monkeypatch) -> None:
calls: list[str] = []
fake_forge_log = types.ModuleType("skyvern.forge.sdk.forge_log")
def fake_setup_logger() -> None:
calls.append("setup")
fake_forge_log.setup_logger = fake_setup_logger
monkeypatch.setattr(cli_bootstrap, "_RUNTIME_LOGGING_CONFIGURED", False)
monkeypatch.setitem(sys.modules, "skyvern.forge.sdk.forge_log", fake_forge_log)
cli_bootstrap.configure_cli_runtime_logging()
cli_bootstrap.configure_cli_runtime_logging()
assert calls == ["setup"]
def test_prepare_cli_runtime_loads_env_before_logger(monkeypatch, tmp_path) -> None:
calls: list[str] = []
project_env = tmp_path / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
project_env.write_text("SKYVERN_API_KEY=project-key\n")
fake_forge_log = types.ModuleType("skyvern.forge.sdk.forge_log")
def fake_setup_logger() -> None:
assert cli_bootstrap.os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
assert cli_bootstrap.os.environ["SKYVERN_API_KEY"] == "project-key"
calls.append("setup")
fake_forge_log.setup_logger = fake_setup_logger
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("HOME", str(tmp_path / "home"))
monkeypatch.delenv("SKYVERN_API_KEY", raising=False)
monkeypatch.delenv(BACKEND_ENV_INTENT_ENV_VAR, raising=False)
monkeypatch.setattr(cli_bootstrap, "_RUNTIME_LOGGING_CONFIGURED", False)
monkeypatch.setitem(sys.modules, "skyvern.forge.sdk.forge_log", fake_forge_log)
assert cli_bootstrap.prepare_cli_runtime(intent=EnvIntent.CLOUD) == project_env
assert calls == ["setup"]

View file

@ -36,14 +36,16 @@ def test_setup_mcp_local_claude_code_uses_local_stdio(monkeypatch) -> None:
def test_init_callback_passes_plain_database_string(monkeypatch) -> None:
calls: list[tuple[bool, str]] = []
calls: list[tuple[bool, str, str | None]] = []
monkeypatch.setattr(
"skyvern.cli.init_command.init_env",
lambda no_postgres=False, database_string="": calls.append((no_postgres, database_string)),
lambda no_postgres=False, database_string="", env_scope=None: calls.append(
(no_postgres, database_string, env_scope)
),
)
result = CliRunner().invoke(cli_app, ["init"])
assert result.exit_code == 0
assert calls == [(False, "")]
assert calls == [(False, "", None)]

View file

@ -1,11 +1,417 @@
from __future__ import annotations
import json
import os
import subprocess
import sys
import types
from pathlib import Path
import pytest
from typer.testing import CliRunner
import skyvern.cli.quickstart as quickstart_module
from skyvern.cli.llm_setup import update_or_add_env_var
from skyvern.utils.env_paths import (
BACKEND_ENV_FILE_ENV_VAR,
BACKEND_ENV_INTENT_ENV_VAR,
EnvIntent,
EnvScope,
backend_env_path_for_scope,
load_backend_env_files,
resolve_backend_env_path,
)
def _set_home(monkeypatch, home: Path) -> None:
home.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HOME", str(home))
def test_backend_env_read_uses_intent_specific_precedence(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
legacy_env = tmp_path / ".env"
global_env = tmp_path / "home" / ".skyvern" / ".env"
global_env.parent.mkdir(parents=True)
legacy_env.write_text("SKYVERN_BASE_URL=http://legacy\n")
global_env.write_text("SKYVERN_BASE_URL=http://global\n")
assert resolve_backend_env_path() == legacy_env
assert resolve_backend_env_path(intent=EnvIntent.SERVER) == legacy_env
assert resolve_backend_env_path(intent=EnvIntent.CLOUD) == global_env
assert backend_env_path_for_scope("cwd") == legacy_env
assert backend_env_path_for_scope("2") == tmp_path / ".skyvern" / ".env"
def test_backend_env_loader_layers_cloud_scopes_and_keeps_server_legacy(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", "GLOBAL_ONLY", "LEGACY_ONLY", BACKEND_ENV_INTENT_ENV_VAR):
monkeypatch.delenv(key, raising=False)
legacy_env = tmp_path / ".env"
project_env = tmp_path / ".skyvern" / ".env"
global_env = tmp_path / "home" / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
global_env.parent.mkdir(parents=True)
legacy_env.write_text("SKYVERN_API_KEY=legacy-key\nLEGACY_ONLY=legacy\n")
global_env.write_text("SKYVERN_API_KEY=global-key\nGLOBAL_ONLY=global\n")
project_env.write_text("SKYVERN_BASE_URL=http://project\n")
assert load_backend_env_files(intent=EnvIntent.CLOUD) == project_env
assert os.environ["SKYVERN_API_KEY"] == "global-key"
assert os.environ["SKYVERN_BASE_URL"] == "http://project"
assert os.environ["GLOBAL_ONLY"] == "global"
assert os.environ["LEGACY_ONLY"] == "legacy"
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", "GLOBAL_ONLY", "LEGACY_ONLY"):
monkeypatch.delenv(key, raising=False)
assert load_backend_env_files(intent=EnvIntent.SERVER) == legacy_env
assert os.environ["SKYVERN_API_KEY"] == "legacy-key"
assert "SKYVERN_BASE_URL" not in os.environ
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.SERVER.value
def test_backend_env_loader_preserves_scope_precedence_across_staged_files(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
for key in ("SKYVERN_API_KEY", "GLOBAL_PROD_ONLY", BACKEND_ENV_INTENT_ENV_VAR):
monkeypatch.delenv(key, raising=False)
project_env = tmp_path / ".skyvern" / ".env"
global_prod_env = tmp_path / "home" / ".skyvern" / ".env.prod"
project_env.parent.mkdir(parents=True)
global_prod_env.parent.mkdir(parents=True)
project_env.write_text("SKYVERN_API_KEY=project-key\n")
global_prod_env.write_text("SKYVERN_API_KEY=global-prod-key\nGLOBAL_PROD_ONLY=yes\n")
assert load_backend_env_files(intent=EnvIntent.CLOUD) == project_env
assert os.environ["SKYVERN_API_KEY"] == "project-key"
assert os.environ["GLOBAL_PROD_ONLY"] == "yes"
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
def test_backend_env_scope_normalization_rejects_unknown_values() -> None:
with pytest.raises(ValueError, match="Choose one of:"):
backend_env_path_for_scope("workspace")
def test_backend_env_loader_preserves_staged_dotenv_precedence(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
for key in ("PORT", "SKYVERN_BASE_URL", BACKEND_ENV_INTENT_ENV_VAR):
monkeypatch.delenv(key, raising=False)
legacy_env = tmp_path / ".env"
legacy_prod_env = tmp_path / ".env.prod"
legacy_env.write_text("PORT=8000\nSKYVERN_BASE_URL=http://server\n")
legacy_prod_env.write_text("PORT=9000\n")
assert load_backend_env_files(intent=EnvIntent.SERVER) == legacy_env
assert os.environ["PORT"] == "9000"
assert os.environ["SKYVERN_BASE_URL"] == "http://server"
def test_server_intent_settings_ignore_cloud_env_files(tmp_path) -> None:
home = tmp_path / "home"
global_env = home / ".skyvern" / ".env"
global_env.parent.mkdir(parents=True)
global_env.write_text("SKYVERN_BASE_URL=http://cloud\nPORT=9001\n")
(tmp_path / ".env").write_text("SKYVERN_BASE_URL=http://server\nPORT=8765\n")
script = """
import json
import os
from skyvern.utils.env_paths import EnvIntent, load_backend_env_files
for key in ("SKYVERN_BASE_URL", "PORT", "SKYVERN_ENV_INTENT"):
os.environ.pop(key, None)
load_backend_env_files(intent=EnvIntent.SERVER)
from skyvern.config import settings
print(json.dumps({"base_url": settings.SKYVERN_BASE_URL, "port": settings.PORT}))
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=tmp_path,
env={**os.environ, "HOME": str(home)},
text=True,
capture_output=True,
check=True,
)
assert json.loads(result.stdout) == {"base_url": "http://server", "port": 8765}
def test_unscoped_settings_import_prefers_legacy_env(tmp_path) -> None:
home = tmp_path / "home"
project_env = tmp_path / ".skyvern" / ".env"
global_env = home / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
global_env.parent.mkdir(parents=True)
(tmp_path / ".env").write_text("SKYVERN_BASE_URL=http://legacy\nPORT=8765\n")
project_env.write_text("SKYVERN_BASE_URL=http://project\nPORT=9001\n")
global_env.write_text("SKYVERN_BASE_URL=http://global\nPORT=9002\n")
script = """
import json
from skyvern.config import settings
print(json.dumps({"base_url": settings.SKYVERN_BASE_URL, "port": settings.PORT}))
"""
env = {**os.environ, "HOME": str(home)}
for key in ("SKYVERN_BASE_URL", "PORT", "SKYVERN_ENV_INTENT", "SKYVERN_ENV_FILE"):
env.pop(key, None)
result = subprocess.run(
[sys.executable, "-c", script],
cwd=tmp_path,
env=env,
text=True,
capture_output=True,
check=True,
)
assert json.loads(result.stdout) == {"base_url": "http://legacy", "port": 8765}
def test_run_commands_import_does_not_initialize_settings_before_server_intent(tmp_path) -> None:
home = tmp_path / "home"
global_env = home / ".skyvern" / ".env"
global_env.parent.mkdir(parents=True)
global_env.write_text("PORT=9001\n")
(tmp_path / ".env").write_text("PORT=8765\n")
script = """
import json
import os
import sys
for key in ("PORT", "SKYVERN_ENV_INTENT"):
os.environ.pop(key, None)
import skyvern.cli.run_commands
config_imported_before_intent = "skyvern.config" in sys.modules
from skyvern.utils.env_paths import EnvIntent, load_backend_env_files
load_backend_env_files(intent=EnvIntent.SERVER)
from skyvern.config import settings
print(json.dumps({"config_imported_before_intent": config_imported_before_intent, "port": settings.PORT}))
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=tmp_path,
env={**os.environ, "HOME": str(home)},
text=True,
capture_output=True,
check=True,
)
assert json.loads(result.stdout) == {"config_imported_before_intent": False, "port": 8765}
def test_run_mcp_prepares_cloud_env_before_starting_mcp(tmp_path, monkeypatch) -> None:
from skyvern import _cli_bootstrap
from skyvern.cli import run_commands
project_env = tmp_path / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
project_env.write_text("SKYVERN_BASE_URL=http://project\nSKYVERN_API_KEY=project-key\n")
(tmp_path / ".env").write_text("SKYVERN_BASE_URL=http://legacy\nSKYVERN_API_KEY=legacy-key\n")
events: list[str] = []
fake_forge_log = types.ModuleType("skyvern.forge.sdk.forge_log")
def fake_setup_logger() -> None:
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
assert os.environ["SKYVERN_BASE_URL"] == "http://project"
events.append("setup_logger")
fake_forge_log.setup_logger = fake_setup_logger
fake_auth = types.ModuleType("skyvern.cli.core.mcp_http_auth")
fake_auth.MCPAPIKeyMiddleware = object
fake_session_manager = types.ModuleType("skyvern.cli.core.session_manager")
fake_session_manager.set_stateless_http_mode = lambda enabled: events.append(f"stateless:{enabled}")
fake_telemetry = types.ModuleType("skyvern.cli.mcp_tools.telemetry")
def fake_configure_mcp_telemetry_runtime(*, server_mode: str, transport: str | None) -> None:
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
assert os.environ["SKYVERN_BASE_URL"] == "http://project"
events.append(f"telemetry:{server_mode}:{transport}")
fake_telemetry.configure_mcp_telemetry_runtime = fake_configure_mcp_telemetry_runtime
fake_mcp_tools = types.ModuleType("skyvern.cli.mcp_tools")
class FakeMCP:
def run(self, *, transport: str, **_: object) -> None:
assert os.environ[BACKEND_ENV_INTENT_ENV_VAR] == EnvIntent.CLOUD.value
assert os.environ["SKYVERN_BASE_URL"] == "http://project"
events.append(f"run:{transport}")
fake_mcp_tools.mcp = FakeMCP()
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("HOME", str(tmp_path / "home"))
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", BACKEND_ENV_INTENT_ENV_VAR):
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(_cli_bootstrap, "_RUNTIME_LOGGING_CONFIGURED", False)
monkeypatch.setattr(run_commands.atexit, "register", lambda _: None)
monkeypatch.setattr(run_commands, "_cleanup_mcp_resources_blocking", lambda: events.append("cleanup"))
monkeypatch.setitem(sys.modules, "skyvern.forge.sdk.forge_log", fake_forge_log)
monkeypatch.setitem(sys.modules, "skyvern.cli.core.mcp_http_auth", fake_auth)
monkeypatch.setitem(sys.modules, "skyvern.cli.core.session_manager", fake_session_manager)
monkeypatch.setitem(sys.modules, "skyvern.cli.mcp_tools", fake_mcp_tools)
monkeypatch.setitem(sys.modules, "skyvern.cli.mcp_tools.telemetry", fake_telemetry)
run_commands.run_mcp()
assert events == [
"setup_logger",
"telemetry:local_cli:stdio",
"stateless:False",
"run:stdio",
"stateless:False",
"cleanup",
]
@pytest.mark.parametrize(
"module_name",
[
"skyvern.cli.workflow",
"skyvern.cli.credential",
"skyvern.cli.schedule_command",
"skyvern.cli.config_command",
"skyvern.cli.block",
"skyvern.cli.setup_commands",
"skyvern.cli.mcp_commands",
"skyvern.cli.init_command",
"skyvern.cli.quickstart",
],
)
def test_cloud_command_import_does_not_initialize_settings_before_cloud_intent(
tmp_path: Path, module_name: str
) -> None:
home = tmp_path / "home"
project_env = tmp_path / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
project_env.write_text("SKYVERN_BASE_URL=http://project\n")
(tmp_path / ".env").write_text("SKYVERN_BASE_URL=http://legacy\n")
script = f"""
import importlib
import json
import os
import sys
for key in ("SKYVERN_BASE_URL", "SKYVERN_ENV_INTENT"):
os.environ.pop(key, None)
importlib.import_module({module_name!r})
config_imported_before_intent = "skyvern.config" in sys.modules
from skyvern.utils.env_paths import EnvIntent, load_backend_env_files
load_backend_env_files(intent=EnvIntent.CLOUD)
from skyvern.config import settings
payload = {{
"config_imported_before_intent": config_imported_before_intent,
"base_url": settings.SKYVERN_BASE_URL,
}}
print(json.dumps(payload))
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=tmp_path,
env={**os.environ, "HOME": str(home)},
text=True,
capture_output=True,
check=True,
)
assert json.loads(result.stdout) == {
"config_imported_before_intent": False,
"base_url": "http://project",
}
def test_setup_credentials_use_cloud_and_local_env_intents(tmp_path) -> None:
home = tmp_path / "home"
project_env = tmp_path / ".skyvern" / ".env"
global_env = home / ".skyvern" / ".env"
project_env.parent.mkdir(parents=True)
global_env.parent.mkdir(parents=True)
(tmp_path / ".env").write_text("SKYVERN_API_KEY=legacy-key\nSKYVERN_BASE_URL=http://legacy\n")
project_env.write_text("SKYVERN_API_KEY=project-key\nSKYVERN_BASE_URL=http://project\n")
global_env.write_text("SKYVERN_API_KEY=global-key\nSKYVERN_BASE_URL=http://global\n")
env = {**os.environ, "HOME": str(home)}
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", "SKYVERN_ENV_INTENT", "SKYVERN_ENV_FILE"):
env.pop(key, None)
remote_script = """
import json
from skyvern.cli.setup_commands import _get_env_credentials
print(json.dumps(dict(zip(("api_key", "base_url"), _get_env_credentials()))))
"""
remote_result = subprocess.run(
[sys.executable, "-c", remote_script],
cwd=tmp_path,
env=env,
text=True,
capture_output=True,
check=True,
)
assert json.loads(remote_result.stdout) == {"api_key": "project-key", "base_url": "http://project"}
local_script = """
import json
from skyvern.cli.setup_commands import _get_local_env_credentials
print(json.dumps(dict(zip(("api_key", "base_url"), _get_local_env_credentials()))))
"""
local_result = subprocess.run(
[sys.executable, "-c", local_script],
cwd=tmp_path,
env=env,
text=True,
capture_output=True,
check=True,
)
assert json.loads(local_result.stdout) == {"api_key": "legacy-key", "base_url": "http://legacy"}
def test_backend_env_write_defaults_are_intent_specific(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
assert resolve_backend_env_path(intent=EnvIntent.CLOUD, for_write=True) == backend_env_path_for_scope(
EnvScope.GLOBAL
)
assert resolve_backend_env_path(intent=EnvIntent.LOCAL, for_write=True) == backend_env_path_for_scope(
EnvScope.PROJECT
)
assert resolve_backend_env_path(intent=EnvIntent.SERVER, for_write=True) == backend_env_path_for_scope(
EnvScope.LEGACY
)
monkeypatch.setenv(BACKEND_ENV_FILE_ENV_VAR, str(tmp_path / "custom.env"))
assert resolve_backend_env_path(intent=EnvIntent.CLOUD, for_write=True) == tmp_path / "custom.env"
def test_update_or_add_env_var_creates_selected_env_parent(tmp_path, monkeypatch) -> None:
monkeypatch.chdir(tmp_path)
_set_home(monkeypatch, tmp_path / "home")
monkeypatch.delenv(BACKEND_ENV_FILE_ENV_VAR, raising=False)
project_env = resolve_backend_env_path(scope=EnvScope.PROJECT, for_write=True)
update_or_add_env_var("SKYVERN_BASE_URL", "http://localhost:8000", env_path=project_env)
assert project_env.exists()
assert not (tmp_path / ".env").exists()
assert "SKYVERN_BASE_URL" in project_env.read_text()
def test_quickstart_without_server_extra_prints_install_paths(monkeypatch) -> None:
@ -267,6 +673,35 @@ def test_quickstart_with_server_extra_preserves_existing_flow(monkeypatch) -> No
]
def test_quickstart_rejects_non_legacy_env_scope_for_server_init(monkeypatch) -> None:
calls = []
monkeypatch.setattr(quickstart_module, "check_docker_compose_file", lambda: False)
monkeypatch.setattr(quickstart_module, "_has_local_quickstart_extra", lambda: True)
monkeypatch.setattr(quickstart_module, "_has_server_quickstart_extra", lambda: True)
monkeypatch.setattr(quickstart_module, "_is_interactive_input", lambda: False)
monkeypatch.setattr(
quickstart_module,
"_run_server_quickstart",
lambda **kwargs: calls.append(kwargs),
)
result = CliRunner().invoke(
quickstart_module.quickstart_app,
[
"--install-type",
"server",
"--env-scope",
"project",
"--server-only",
],
)
assert result.exit_code == 1
assert "Self-hosted local server setup writes ./.env" in result.output
assert calls == []
def test_quickstart_docker_compose_bypasses_server_extra_guard(monkeypatch) -> None:
calls = []
@ -301,7 +736,7 @@ async def test_start_services_without_frontend_runtime_starts_backend_only(monke
return None
monkeypatch.setattr(utils, "resolve_frontend_env_path", lambda: None)
monkeypatch.setattr(utils, "resolve_backend_env_path", lambda: Path(".env"))
monkeypatch.setattr(utils, "resolve_backend_env_path", lambda **_kwargs: Path(".env"))
monkeypatch.setattr(utils.asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
monkeypatch.setattr(utils.asyncio, "sleep", fake_sleep)
monkeypatch.setattr(utils, "capture_setup_event", lambda *args, **kwargs: None)

View file

@ -102,6 +102,61 @@ def test_setup_hermes_local_fails_without_base_url(hermes_home: Path, monkeypatc
assert "SKYVERN_BASE_URL" in result.output
def test_setup_hermes_local_uses_legacy_env_when_cloud_scope_exists(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr("skyvern.cli.setup_commands.Path.home", lambda: tmp_path)
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", "SKYVERN_ENV_FILE", "SKYVERN_ENV_INTENT"):
monkeypatch.delenv(key, raising=False)
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_save_yaml_config(hermes_home / "config.yaml", {})
local_env = tmp_path / ".env"
cloud_env = tmp_path / ".skyvern" / ".env"
cloud_env.parent.mkdir(parents=True)
local_env.write_text("SKYVERN_API_KEY=local-key\nSKYVERN_BASE_URL=http://localhost:8000\n")
cloud_env.write_text("SKYVERN_API_KEY=cloud-key\nSKYVERN_BASE_URL=https://api.skyvern.com\n")
result = runner.invoke(setup_app, ["hermes", "--local", "--yes"])
assert result.exit_code == 0, result.output
data = _load_yaml_config(hermes_home / "config.yaml")
assert data is not None
entry = data["mcp_servers"]["skyvern"]
assert entry["env"]["SKYVERN_API_KEY"] == "local-key"
assert entry["env"]["SKYVERN_BASE_URL"] == "http://localhost:8000"
def test_setup_hermes_local_does_not_pair_cloud_key_with_legacy_url(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.setattr("skyvern.cli.setup_commands.Path.home", lambda: tmp_path)
for key in ("SKYVERN_API_KEY", "SKYVERN_BASE_URL", "SKYVERN_ENV_FILE", "SKYVERN_ENV_INTENT"):
monkeypatch.delenv(key, raising=False)
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
_save_yaml_config(hermes_home / "config.yaml", {})
local_env = tmp_path / ".env"
cloud_env = tmp_path / ".skyvern" / ".env"
cloud_env.parent.mkdir(parents=True)
local_env.write_text("SKYVERN_BASE_URL=http://localhost:8000\n")
cloud_env.write_text("SKYVERN_API_KEY=cloud-key\nSKYVERN_BASE_URL=https://api.skyvern.com\n")
result = runner.invoke(setup_app, ["hermes", "--local", "--yes"])
assert result.exit_code == 1
assert "No API key found" in result.output
data = _load_yaml_config(hermes_home / "config.yaml")
assert data is not None
assert "mcp_servers" not in data
def test_setup_hermes_dry_run_masks_secrets(hermes_home: Path) -> None:
"""Dry-run output does not contain raw API keys."""
result = runner.invoke(setup_app, ["hermes", "--dry-run"])