Automatically run a server in a subprocess if no API URL is specified

.github/workflows/benchmarks.yaml

@@ -105,8 +105,8 @@ jobs:
           # characters with an underscore
           sanitized_uniquename="${uniquename//[^a-zA-Z0-9_\-]/_}"
 
-          PREFECT_API_URL="http://127.0.0.1:4200/api"
-          python benches \
+          PREFECT_API_URL="http://127.0.0.1:4200/api" \
+          python -m benches \
           --ignore=benches/bench_import.py \
           --timeout=180 \
           --benchmark-save="${sanitized_uniquename}" \

benches/bench_flows.py

@@ -55,7 +55,10 @@ def benchmark_flow():
         for _ in range(num_tasks):
             test_task()
 
-    benchmark(benchmark_flow)
+    if num_tasks > 100:
+        benchmark.pedantic(benchmark_flow)
+    else:
+        benchmark(benchmark_flow)
 
 
 @pytest.mark.parametrize("num_tasks", [10, 50, 100, 250])
@@ -68,7 +71,10 @@ async def benchmark_flow():
             for _ in range(num_tasks):
                 tg.start_soon(test_task)
 
-    benchmark(anyio.run, benchmark_flow)
+    if num_tasks > 100:
+        benchmark.pedantic(anyio.run, (benchmark_flow,))
+    else:
+        benchmark(anyio.run, benchmark_flow)
 
 
 @pytest.mark.parametrize("num_flows", [5, 10, 20])

docs/3.0rc/api-ref/rest-api/server/schema.json

@@ -22271,6 +22271,16 @@
                         "title": "Prefect Server Csrf Token Expiration",
                         "default": "PT1H"
                     },
+                    "PREFECT_SERVER_ALLOW_EPHEMERAL_MODE": {
+                        "type": "boolean",
+                        "title": "Prefect Server Allow Ephemeral Mode",
+                        "default": false
+                    },
+                    "PREFECT_SERVER_EPHEMERAL_STARTUP_TIMEOUT_SECONDS": {
+                        "type": "integer",
+                        "title": "Prefect Server Ephemeral Startup Timeout Seconds",
+                        "default": 10
+                    },
                     "PREFECT_UI_ENABLED": {
                         "type": "boolean",
                         "title": "Prefect Ui Enabled",

src/prefect/blocks/core.py

@@ -902,7 +902,9 @@ class Custom(Block):
             loaded_block.save("my-custom-message", overwrite=True)
             ```
         """
-        block_document, block_document_name = await cls._get_block_document(name)
+        block_document, block_document_name = await cls._get_block_document(
+            name, client=client
+        )
 
         return cls._load_from_block_document(block_document, validate=validate)
 

src/prefect/cli/profile.py

@@ -18,6 +18,7 @@
 from prefect.cli._utilities import exit_with_error, exit_with_success
 from prefect.cli.cloud import CloudUnauthorizedError, get_cloud_client
 from prefect.cli.root import app, is_interactive
+from prefect.client.base import determine_server_type
 from prefect.client.orchestration import ServerType, get_client
 from prefect.context import use_profile
 from prefect.exceptions import ObjectNotFound
@@ -138,6 +139,13 @@ async def use(name: str):
                 " in ephemeral mode."
             ),
         ),
+        ConnectionStatus.UNCONFIGURED: (
+            exit_with_error,
+            (
+                f"Prefect server URL not configured using profile {name!r} - please"
+                " configure the server URL or enable ephemeral mode."
+            ),
+        ),
         ConnectionStatus.INVALID_API: (
             exit_with_error,
             "Error connecting to Prefect API URL",
@@ -350,6 +358,7 @@ class ConnectionStatus(AutoEnum):
     CLOUD_UNAUTHORIZED = AutoEnum.auto()
     SERVER_CONNECTED = AutoEnum.auto()
     SERVER_ERROR = AutoEnum.auto()
+    UNCONFIGURED = AutoEnum.auto()
     EPHEMERAL = AutoEnum.auto()
     INVALID_API = AutoEnum.auto()
 
@@ -373,14 +382,16 @@ async def check_server_connection():
         try:
             # inform the user if Prefect API endpoints exist, but there are
             # connection issues
+            server_type = determine_server_type()
+            if server_type == ServerType.EPHEMERAL:
+                return ConnectionStatus.EPHEMERAL
+            elif server_type == ServerType.UNCONFIGURED:
+                return ConnectionStatus.UNCONFIGURED
             client = get_client(httpx_settings=httpx_settings)
             async with client:
                 connect_error = await client.api_healthcheck()
             if connect_error is not None:
                 return ConnectionStatus.SERVER_ERROR
-            elif client.server_type == ServerType.EPHEMERAL:
-                # if the client is using an ephemeral Prefect app, inform the user
-                return ConnectionStatus.EPHEMERAL
             else:
                 return ConnectionStatus.SERVER_CONNECTED
         except Exception:
@@ -390,6 +401,13 @@ async def check_server_connection():
     except TypeError:
         # if no Prefect API URL has been set, httpx will throw a TypeError
         try:
+            # try to connect with the client anyway, it will likely use an
+            # ephemeral Prefect instance
+            server_type = determine_server_type()
+            if server_type == ServerType.EPHEMERAL:
+                return ConnectionStatus.EPHEMERAL
+            elif server_type == ServerType.UNCONFIGURED:
+                return ConnectionStatus.UNCONFIGURED
             client = get_client(httpx_settings=httpx_settings)
             if client.server_type == ServerType.EPHEMERAL:
                 return ConnectionStatus.EPHEMERAL

src/prefect/cli/root.py

@@ -17,6 +17,7 @@
 import prefect.settings
 from prefect.cli._types import PrefectTyper, SettingsOption
 from prefect.cli._utilities import with_cli_exception_handling
+from prefect.client.base import determine_server_type
 from prefect.client.constants import SERVER_API_VERSION
 from prefect.client.orchestration import ServerType
 from prefect.logging.configuration import setup_logging
@@ -117,16 +118,7 @@ async def version(
         "OS/Arch": f"{sys.platform}/{platform.machine()}",
         "Profile": prefect.context.get_settings_context().profile.name,
     }
-
-    server_type: str
-
-    try:
-        # We do not context manage the client because when using an ephemeral app we do not
-        # want to create the database or run migrations
-        client = prefect.get_client()
-        server_type = client.server_type.value
-    except Exception:
-        server_type = "<client error>"
+    server_type = determine_server_type()
 
     version_info["Server type"] = server_type.lower()
 

src/prefect/client/base.py

@@ -35,10 +35,14 @@
 from prefect.exceptions import PrefectHTTPStatusError
 from prefect.logging import get_logger
 from prefect.settings import (
+    PREFECT_API_URL,
     PREFECT_CLIENT_MAX_RETRIES,
     PREFECT_CLIENT_RETRY_EXTRA_CODES,
     PREFECT_CLIENT_RETRY_JITTER_FACTOR,
+    PREFECT_CLOUD_API_URL,
+    PREFECT_SERVER_ALLOW_EPHEMERAL_MODE,
 )
+from prefect.utilities.collections import AutoEnum
 from prefect.utilities.math import bounded_poisson_interval, clamped_poisson_interval
 
 # Datastores for lifespan management, keys should be a tuple of thread and app
@@ -637,3 +641,33 @@ def __init__(
         )
 
     pass
+
+
+class ServerType(AutoEnum):
+    EPHEMERAL = AutoEnum.auto()
+    SERVER = AutoEnum.auto()
+    CLOUD = AutoEnum.auto()
+    UNCONFIGURED = AutoEnum.auto()
+
+
+def determine_server_type() -> ServerType:
+    """
+    Determine the server type based on the current settings.
+
+    Returns:
+        - `ServerType.EPHEMERAL` if the ephemeral server is enabled
+        - `ServerType.SERVER` if a API URL is configured and it is not a cloud URL
+        - `ServerType.CLOUD` if an API URL is configured and it is a cloud URL
+        - `ServerType.UNCONFIGURED` if no API URL is configured and ephemeral mode is
+            not enabled
+    """
+    api_url = PREFECT_API_URL.value()
+    if api_url is None:
+        if PREFECT_SERVER_ALLOW_EPHEMERAL_MODE.value():
+            return ServerType.EPHEMERAL
+        else:
+            return ServerType.UNCONFIGURED
+    if api_url.startswith(PREFECT_CLOUD_API_URL.value()):
+        return ServerType.CLOUD
+    else:
+        return ServerType.SERVER

src/prefect/client/orchestration.py

@@ -132,9 +132,9 @@
     PREFECT_API_URL,
     PREFECT_CLIENT_CSRF_SUPPORT_ENABLED,
     PREFECT_CLOUD_API_URL,
+    PREFECT_SERVER_ALLOW_EPHEMERAL_MODE,
     PREFECT_UNIT_TEST_MODE,
 )
-from prefect.utilities.collections import AutoEnum
 
 if TYPE_CHECKING:
     from prefect.flows import Flow as FlowObject
@@ -145,19 +145,14 @@
     PrefectHttpxAsyncClient,
     PrefectHttpxSyncClient,
     PrefectHttpxSyncEphemeralClient,
+    ServerType,
     app_lifespan_context,
 )
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
 
-class ServerType(AutoEnum):
-    EPHEMERAL = AutoEnum.auto()
-    SERVER = AutoEnum.auto()
-    CLOUD = AutoEnum.auto()
-
-
 @overload
 def get_client(
     httpx_settings: Optional[Dict[str, Any]] = None, sync_client: Literal[False] = False
@@ -194,8 +189,6 @@ def get_client(
     """
     import prefect.context
 
-    settings_ctx = prefect.context.get_settings_context()
-
     # try to load clients from a client context, if possible
     # only load clients that match the provided config / loop
     try:
@@ -217,24 +210,36 @@ def get_client(
                 return client_ctx.client
 
     api = PREFECT_API_URL.value()
+    server_type = None
 
-    if not api:
+    if not api and PREFECT_SERVER_ALLOW_EPHEMERAL_MODE:
         # create an ephemeral API if none was provided
-        from prefect.server.api.server import create_app
+        from prefect.server.api.server import SubprocessASGIServer
 
-        api = create_app(settings_ctx.settings, ephemeral=True)
+        server = SubprocessASGIServer()
+        server.start()
+        assert server.server_process is not None, "Server process did not start"
+
+        api = f"{server.address()}/api"
+        server_type = ServerType.EPHEMERAL
+    elif not api and not PREFECT_SERVER_ALLOW_EPHEMERAL_MODE:
+        raise ValueError(
+            "No Prefect API URL provided. Please set PREFECT_API_URL to the address of a running Prefect server."
+        )
 
     if sync_client:
         return SyncPrefectClient(
             api,
             api_key=PREFECT_API_KEY.value(),
             httpx_settings=httpx_settings,
+            server_type=server_type,
         )
     else:
         return PrefectClient(
             api,
             api_key=PREFECT_API_KEY.value(),
             httpx_settings=httpx_settings,
+            server_type=server_type,
         )
 
 
@@ -271,6 +276,7 @@ def __init__(
         api_key: Optional[str] = None,
         api_version: Optional[str] = None,
         httpx_settings: Optional[Dict[str, Any]] = None,
+        server_type: Optional[ServerType] = None,
     ) -> None:
         httpx_settings = httpx_settings.copy() if httpx_settings else {}
         httpx_settings.setdefault("headers", {})
@@ -333,11 +339,14 @@ def __init__(
             # client will use a standard HTTP/1.1 connection instead.
             httpx_settings.setdefault("http2", PREFECT_API_ENABLE_HTTP2.value())
 
-            self.server_type = (
-                ServerType.CLOUD
-                if api.startswith(PREFECT_CLOUD_API_URL.value())
-                else ServerType.SERVER
-            )
+            if server_type:
+                self.server_type = server_type
+            else:
+                self.server_type = (
+                    ServerType.CLOUD
+                    if api.startswith(PREFECT_CLOUD_API_URL.value())
+                    else ServerType.SERVER
+                )
 
         # Connect to an in-process application
         elif isinstance(api, ASGIApp):
@@ -3386,6 +3395,7 @@ def __init__(
         api_key: Optional[str] = None,
         api_version: Optional[str] = None,
         httpx_settings: Optional[Dict[str, Any]] = None,
+        server_type: Optional[ServerType] = None,
     ) -> None:
         httpx_settings = httpx_settings.copy() if httpx_settings else {}
         httpx_settings.setdefault("headers", {})
@@ -3444,11 +3454,14 @@ def __init__(
             # client will use a standard HTTP/1.1 connection instead.
             httpx_settings.setdefault("http2", PREFECT_API_ENABLE_HTTP2.value())
 
-            self.server_type = (
-                ServerType.CLOUD
-                if api.startswith(PREFECT_CLOUD_API_URL.value())
-                else ServerType.SERVER
-            )
+            if server_type:
+                self.server_type = server_type
+            else:
+                self.server_type = (
+                    ServerType.CLOUD
+                    if api.startswith(PREFECT_CLOUD_API_URL.value())
+                    else ServerType.SERVER
+                )
 
         # Connect to an in-process application
         elif isinstance(api, ASGIApp):
@@ -4062,6 +4075,33 @@ def read_deployment(
                 raise
         return DeploymentResponse.model_validate(response.json())
 
+    def read_deployment_by_name(
+        self,
+        name: str,
+    ) -> DeploymentResponse:
+        """
+        Query the Prefect API for a deployment by name.
+
+        Args:
+            name: A deployed flow's name: <FLOW_NAME>/<DEPLOYMENT_NAME>
+
+        Raises:
+            prefect.exceptions.ObjectNotFound: If request returns 404
+            httpx.RequestError: If request fails
+
+        Returns:
+            a Deployment model representation of the deployment
+        """
+        try:
+            response = self._client.get(f"/deployments/name/{name}")
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == status.HTTP_404_NOT_FOUND:
+                raise prefect.exceptions.ObjectNotFound(http_exc=e) from e
+            else:
+                raise
+
+        return DeploymentResponse.model_validate(response.json())
+
     def create_artifact(
         self,
         artifact: ArtifactCreate,

src/prefect/events/worker.py

@@ -17,7 +17,6 @@
     EventsClient,
     NullEventsClient,
     PrefectCloudEventsClient,
-    PrefectEphemeralEventsClient,
     PrefectEventsClient,
 )
 from .related import related_resources_from_run_context
@@ -97,7 +96,15 @@ def instance(
             elif should_emit_events_to_running_server():
                 client_type = PrefectEventsClient
             elif should_emit_events_to_ephemeral_server():
-                client_type = PrefectEphemeralEventsClient
+                # create an ephemeral API if none was provided
+                from prefect.server.api.server import SubprocessASGIServer
+
+                server = SubprocessASGIServer()
+                server.start()
+                assert server.server_process is not None, "Server process did not start"
+
+                client_kwargs = {"api_url": f"{server.address()}/api"}
+                client_type = PrefectEventsClient
             else:
                 client_type = NullEventsClient