🌿 Fern Regeneration -- March 13, 2026

pyproject.toml

@@ -4,7 +4,7 @@ dynamic = ["version"]
 
 [tool.poetry]
 name = "pipedream"
-version = "1.1.5"
+version = "1.1.6"
 description = ""
 readme = "README.md"
 authors = []

src/pipedream/core/client_wrapper.py

@@ -27,10 +27,10 @@ def __init__(
 
     def get_headers(self) -> typing.Dict[str, str]:
         headers: typing.Dict[str, str] = {
-            "User-Agent": "pipedream/1.1.5",
+            "User-Agent": "pipedream/1.1.6",
             "X-Fern-Language": "Python",
             "X-Fern-SDK-Name": "pipedream",
-            "X-Fern-SDK-Version": "1.1.5",
+            "X-Fern-SDK-Version": "1.1.6",
             **(self.get_custom_headers() or {}),
         }
         if self._project_environment is not None:

src/pipedream/environment.py

@@ -4,6 +4,4 @@
 
 
 class PipedreamEnvironment(enum.Enum):
-    CANARY = "https://api2.pipedream.com"
-    DEV = "https://api.${DEV_NAMESPACE}.gkes.pipedream.net"
     PROD = "https://api.pipedream.com"

src/pipedream/pipedream.py

@@ -5,13 +5,14 @@
     AsyncClient,
     Client,
 )
-from .environment import PipedreamEnvironment
 from .types.project_environment import ProjectEnvironment
 from .workflows.client import (
     AsyncWorkflowsClient,
     WorkflowsClient,
 )
 
+_PROD_BASE_URL = "https://api.pipedream.com"
+
 
 class Pipedream(Client):
 
@@ -26,37 +27,37 @@ def __init__(
             "PIPEDREAM_PROJECT_ENVIRONMENT",
             "production",
         ),
-        environment: PipedreamEnvironment = PipedreamEnvironment.PROD,
+        base_url: Optional[str] = None,
         workflow_domain: Optional[str] = None,
         timeout: Optional[float] = None,
     ):
         if not project_id:
             raise ValueError("Project ID is required")
 
+        resolved_base_url = base_url or os.environ.get("PIPEDREAM_BASE_URL") or _PROD_BASE_URL
+        resolved_workflow_domain = workflow_domain or os.environ.get("PIPEDREAM_WORKFLOW_DOMAIN") or "m.pipedream.net"
+
         if access_token:
             super().__init__(
-                base_url=_get_base_url(environment),
+                base_url=resolved_base_url,
                 project_environment=project_environment,
                 project_id=project_id,
                 token=(lambda: access_token),
                 timeout=timeout,
             )
         else:
             super().__init__(
-                base_url=_get_base_url(environment),
+                base_url=resolved_base_url,
                 project_environment=project_environment,
                 project_id=project_id,
                 client_id=client_id,
                 client_secret=client_secret,
                 timeout=timeout,
             )
 
-        if not workflow_domain:
-            workflow_domain = _get_default_workflow_domain(environment)
-
         self.workflows = WorkflowsClient(
             client_wrapper=self._client_wrapper,
-            workflow_domain=workflow_domain,
+            workflow_domain=resolved_workflow_domain,
         )
 
     @property
@@ -80,37 +81,37 @@ def __init__(
             "PIPEDREAM_PROJECT_ENVIRONMENT",
             "production",
         ),
-        environment: PipedreamEnvironment = PipedreamEnvironment.PROD,
+        base_url: Optional[str] = None,
         workflow_domain: Optional[str] = None,
         timeout: Optional[float] = None,
     ):
         if not project_id:
             raise ValueError("Project ID is required")
 
+        resolved_base_url = base_url or os.environ.get("PIPEDREAM_BASE_URL") or _PROD_BASE_URL
+        resolved_workflow_domain = workflow_domain or os.environ.get("PIPEDREAM_WORKFLOW_DOMAIN") or "m.pipedream.net"
+
         if access_token:
             super().__init__(
-                base_url=_get_base_url(environment),
+                base_url=resolved_base_url,
                 project_environment=project_environment,
                 project_id=project_id,
                 token=(lambda: access_token),
                 timeout=timeout,
             )
         else:
             super().__init__(
-                base_url=_get_base_url(environment),
+                base_url=resolved_base_url,
                 project_environment=project_environment,
                 project_id=project_id,
                 client_id=client_id,
                 client_secret=client_secret,
                 timeout=timeout,
             )
 
-        if not workflow_domain:
-            workflow_domain = _get_default_workflow_domain(environment)
-
         self.workflows = AsyncWorkflowsClient(
             client_wrapper=self._client_wrapper,
-            workflow_domain=workflow_domain,
+            workflow_domain=resolved_workflow_domain,
         )
 
     @property
@@ -119,19 +120,3 @@ def raw_access_token(self) -> Optional[str]:
         Returns an access token that can be used to authenticate API requests
         """
         return self._client_wrapper._get_token()
-
-
-def _get_base_url(environment: PipedreamEnvironment) -> str:
-    """
-    Returns the base URL for the given environment.
-    """
-    return os.path.expandvars(environment.value)
-
-
-def _get_default_workflow_domain(environment: PipedreamEnvironment) -> str:
-    """
-    Returns the default workflow domain.
-    """
-    if environment == PipedreamEnvironment.DEV:
-        return "m.d.pipedream.net"
-    return "m.pipedream.net"