From 695963f33b6d4d48a73f5047281f655900fc4247 Mon Sep 17 00:00:00 2001
From: Kayvan Zahiri <123409429+Kayvan-Zahiri@users.noreply.github.com>
Date: Sun, 18 Oct 2026 10:54:19 -0700
Subject: [PATCH] fix(workflows): route invocations to the workflow host

---
 .fernignore                           |  1 +
 src/pipedream/workflows/client.py     |  7 +--
 tests/custom/test_workflow_routing.py | 74 +++++++++++++++++++++++++++
 3 files changed, 79 insertions(+), 3 deletions(-)
 create mode 100644 tests/custom/test_workflow_routing.py

diff --git a/.fernignore b/.fernignore
index bca5c43..624c880 100644
--- a/.fernignore
+++ b/.fernignore
@@ -28,6 +28,7 @@ src/pipedream/proxy/raw_client.py
 # Custom Workflow files
 src/pipedream/workflows/__init__.py
 src/pipedream/workflows/client.py
+tests/custom/test_workflow_routing.py
 
 # Project-level documentation for AI assistants
 CLAUDE.md
diff --git a/src/pipedream/workflows/client.py b/src/pipedream/workflows/client.py
index 1f31604..3dba320 100644
--- a/src/pipedream/workflows/client.py
+++ b/src/pipedream/workflows/client.py
@@ -3,7 +3,6 @@
 import typing
 
 import httpx
-
 from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
 from ..core.request_options import RequestOptions
 
@@ -134,7 +133,8 @@ def invoke(
         request_headers = _prepare_headers(headers, auth_type)
 
         response = self._client_wrapper.httpx_client.request(
-            workflow_url,
+            None,
+            base_url=workflow_url,
             method=method,
             data=body,
             json=body,
@@ -238,7 +238,8 @@ async def invoke(
         request_headers = _prepare_headers(headers, auth_type)
 
         response = await self._client_wrapper.httpx_client.request(
-            workflow_url,
+            None,
+            base_url=workflow_url,
             method=method,
             data=body,
             json=body,
diff --git a/tests/custom/test_workflow_routing.py b/tests/custom/test_workflow_routing.py
new file mode 100644
index 0000000..4266cd7
--- /dev/null
+++ b/tests/custom/test_workflow_routing.py
@@ -0,0 +1,74 @@
+import httpx
+import pytest
+
+from pipedream import AsyncPipedream, Pipedream
+
+TARGETS = [
+    ("https://workflow.example.com/run?mode=test", None, "https://workflow.example.com/run?mode=test"),
+    ("http://localhost:8080/run", None, "http://localhost:8080/run"),
+    ("workflow.example.com/run", None, "https://workflow.example.com/run"),
+    ("en123", None, "https://en123.m.pipedream.net/"),
+    ("en123", "workflows.example.com", "https://en123.workflows.example.com/"),
+]
+
+
+@pytest.mark.parametrize("target,domain,expected", TARGETS)
+@pytest.mark.parametrize("external_user", [False, True])
+def test_workflow_uses_workflow_host(target: str, domain: str | None, expected: str, external_user: bool) -> None:
+    seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return httpx.Response(200, json={"ok": True})
+
+    with httpx.Client(transport=httpx.MockTransport(handler)) as transport:
+        client = Pipedream(
+            access_token="test-token",
+            project_id="test",
+            base_url="https://api.example.com/prefix",
+            workflow_domain=domain,
+            httpx_client=transport,
+        )
+        if external_user:
+            response = client.workflows.invoke_for_external_user(target, external_user_id="user-123")
+        else:
+            response = client.workflows.invoke(target)
+
+    assert response.json() == {"ok": True}
+    assert len(seen) == 1
+    assert str(seen[0].url) == expected
+    assert seen[0].headers["authorization"] == ("Bearer test-token" if external_user else "")
+    if external_user:
+        assert seen[0].headers["x-pd-external-user-id"] == "user-123"
+
+
+@pytest.mark.parametrize("target,domain,expected", TARGETS)
+@pytest.mark.parametrize("external_user", [False, True])
+async def test_async_workflow_uses_workflow_host(
+    target: str, domain: str | None, expected: str, external_user: bool
+) -> None:
+    seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append(request)
+        return httpx.Response(200, json={"ok": True})
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as transport:
+        client = AsyncPipedream(
+            access_token="test-token",
+            project_id="test",
+            base_url="https://api.example.com/prefix",
+            workflow_domain=domain,
+            httpx_client=transport,
+        )
+        if external_user:
+            response = await client.workflows.invoke_for_external_user(target, external_user_id="user-123")
+        else:
+            response = await client.workflows.invoke(target)
+
+    assert response.json() == {"ok": True}
+    assert len(seen) == 1
+    assert str(seen[0].url) == expected
+    assert seen[0].headers["authorization"] == ("Bearer test-token" if external_user else "")
+    if external_user:
+        assert seen[0].headers["x-pd-external-user-id"] == "user-123"