fix: add import path support for python sources

docs/source_architecture.md

@@ -77,7 +77,7 @@ McpSourceConnection
 For Python, the provider is simpler:
 
 ```text
-PythonSourceConfig(module, registry)
+PythonSourceConfig(path, module, registry)
   -> import module
   -> load NodeSpec registry
   -> qualify specs under source id

docs/superpowers/specs/2026-06-11-python-source-provider.md

@@ -56,6 +56,7 @@ Start with static config only:
 {
   "kind": "python",
   "id": "local.ops",
+  "path": ".",
   "module": "my_project.workflow_ops",
   "registry": "registry"
 }
@@ -119,7 +120,7 @@ Implemented:
 
 - `wf_config` accepts `server.sources[]` entries with `kind: "python"`.
 - `wf_sources_python` loads trusted in-process `NodeSpec` registries from
-  `module:registry`.
+  `path` plus `module:registry`.
 - `wf_server.config` composes Python sources into local/static servers.
 - Capability listing/calling works over JSON-RPC.
 

examples/python_source_ops.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+from wf_authoring import node
+
+
+class EchoInput(BaseModel):
+    text: str
+
+
+class EchoOutput(BaseModel):
+    echoed: str
+
+
+@node(name="echo")
+def echo(payload: EchoInput) -> EchoOutput:
+    """Return the submitted text through a project-local Python source."""
+    return EchoOutput(echoed=payload.text)
+
+
+@node(name="authoring.upper")
+def upper(payload: EchoInput) -> EchoOutput:
+    """Upper-case text while exercising authoring-name qualification."""
+    return EchoOutput(echoed=payload.text.upper())
+
+
+registry = [echo, upper]

src/wf_config/loader.py

@@ -5,7 +5,9 @@ from pathlib import Path
 
 from .models import (
     FilesystemStoreConfig,
+    PythonSourceConfig,
     ServerStoresConfig,
+    SourceConfig,
     StoreConfig,
     WorkflowConfigFile,
 )
@@ -49,6 +51,10 @@ def _resolve_store_paths(
                 update={
                     "store": _resolve_store(server.store, base_dir=base_dir),
                     "stores": resolved_stores,
+                    "sources": _resolve_source_paths(
+                        server.sources,
+                        base_dir=base_dir,
+                    ),
                 }
             )
         }
@@ -63,3 +69,19 @@ def _resolve_store(
     if isinstance(store, FilesystemStoreConfig) and not store.root.is_absolute():
         return store.model_copy(update={"root": (base_dir / store.root).resolve()})
     return store
+
+
+def _resolve_source_paths(
+    sources: list[SourceConfig],
+    *,
+    base_dir: Path,
+) -> list[SourceConfig]:
+    resolved: list[SourceConfig] = []
+    for source in sources:
+        if isinstance(source, PythonSourceConfig) and not source.path.is_absolute():
+            resolved.append(
+                source.model_copy(update={"path": (base_dir / source.path).resolve()})
+            )
+        else:
+            resolved.append(source)
+    return resolved

src/wf_config/models.py

@@ -104,6 +104,7 @@ class PythonSourceConfig(WorkflowConfigModel):
     kind: Literal["python"] = "python"
     id: str
     enabled: bool = True
+    path: Path = Path(".")
     module: str = Field(min_length=1)
     registry: str = Field(default="registry", min_length=1)
 

src/wf_server/config.py

@@ -20,6 +20,7 @@ def _python_sources(config: WorkflowConfigFile) -> dict[str, CapabilitySource]:
     return {
         source.id: load_python_source(
             source_id=source.id,
+            path=source.path,
             module=source.module,
             registry=source.registry,
             enabled=source.enabled,

src/wf_sources_python/loader.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import sys
 from collections.abc import Mapping, Sequence
 from importlib import import_module
+from pathlib import Path
 from typing import Any, Protocol
 
 from wf_authoring import NodeSpec, node
@@ -15,6 +17,7 @@ from wf_platform import (
 
 class PythonSourceConfigLike(Protocol):
     id: str
+    path: Path
     module: str
     registry: str
     enabled: bool
@@ -26,8 +29,11 @@ def load_python_source(
     module: str,
     registry: str = "registry",
     enabled: bool = True,
+    path: str | Path | None = None,
 ) -> CapabilitySource:
     """Load a trusted in-process Python source from a module registry object."""
+    if path is not None:
+        _ensure_import_path(Path(path))
     module_obj = import_module(module)
     if not hasattr(module_obj, registry):
         raise ValueError(f"missing registry object {registry!r} in module {module!r}")
@@ -59,6 +65,7 @@ def load_python_source(
 def python_capability_source(config: PythonSourceConfigLike) -> CapabilitySource:
     return load_python_source(
         source_id=config.id,
+        path=config.path,
         module=config.module,
         registry=config.registry,
         enabled=config.enabled,
@@ -83,6 +90,13 @@ def _coerce_specs(raw_registry: object) -> list[NodeSpec[Any, Any]]:
     return specs
 
 
+def _ensure_import_path(path: Path) -> None:
+    """Keep configured trusted source roots importable for deferred imports."""
+    resolved = str(path.resolve())
+    if resolved not in sys.path:
+        sys.path.insert(0, resolved)
+
+
 def _qualify_spec(source_id: str, spec: NodeSpec[Any, Any]) -> NodeSpec[Any, Any]:
     local_name = spec.name.removeprefix("authoring.")
     if local_name.startswith(f"{source_id}."):

tests/wf_config/test_config_models.py

@@ -406,10 +406,39 @@ def test_workflow_config_parses_python_source() -> None:
     source = config.server.sources[0]
     assert source.kind == "python"
     assert source.id == "local.ops"
+    assert source.path == Path(".")
     assert source.module == "tests.fixtures.python_source_ops"
     assert source.registry == "registry"
 
 
+def test_load_workflow_config_resolves_python_source_path(tmp_path: Path) -> None:
+    config_path = tmp_path / "wf.json"
+    config_path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "server": {
+                    "sources": [
+                        {
+                            "kind": "python",
+                            "id": "local.ops",
+                            "path": "src",
+                            "module": "project_ops",
+                        }
+                    ]
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    config = load_workflow_config(config_path)
+
+    source = config.server.sources[0]
+    assert source.kind == "python"
+    assert source.path == (tmp_path / "src").resolve()
+
+
 def test_server_config_resolves_missing_role_stores_to_default_store() -> None:
     config = WorkflowConfigFile.model_validate(
         {

tests/wf_sources_python/test_loader.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 import pytest
 
 from wf_sources_python import load_python_source
@@ -34,6 +36,41 @@ def test_load_python_source_from_callable_registry() -> None:
     }
 
 
+def test_load_python_source_uses_configured_import_path(tmp_path: Path) -> None:
+    source_root = tmp_path / "source_root"
+    source_root.mkdir()
+    (source_root / "project_ops.py").write_text(
+        """
+from __future__ import annotations
+
+from pydantic import BaseModel
+from wf_authoring import node
+
+class EchoInput(BaseModel):
+    text: str
+
+class EchoOutput(BaseModel):
+    echoed: str
+
+@node(name="echo")
+def echo(payload: EchoInput) -> EchoOutput:
+    return EchoOutput(echoed=payload.text)
+
+registry = [echo]
+""",
+        encoding="utf-8",
+    )
+
+    source = load_python_source(
+        source_id="local.ops",
+        path=source_root,
+        module="project_ops",
+        registry="registry",
+    )
+
+    assert set(source.capabilities.node_specs) == {"local.ops.echo"}
+
+
 def test_load_python_source_propagates_enabled_flag() -> None:
     source = load_python_source(
         source_id="local.ops",