feat: compose python sources in workflow server

src/wf_server/config.py

@@ -2,7 +2,8 @@ from __future__ import annotations
 
 from pathlib import Path
 
-from wf_config import FilesystemStoreConfig, WorkflowConfigFile
+from wf_config import FilesystemStoreConfig, PythonSourceConfig, WorkflowConfigFile
+from wf_platform import CapabilitySource
 
 from .context import WorkflowServer, build_local_static_workflow_server
 
@@ -13,6 +14,21 @@ def _has_mcp_sources(config: WorkflowConfigFile) -> bool:
     )
 
 
+def _python_sources(config: WorkflowConfigFile) -> dict[str, CapabilitySource]:
+    from wf_sources_python import load_python_source
+
+    return {
+        source.id: load_python_source(
+            source_id=source.id,
+            module=source.module,
+            registry=source.registry,
+            enabled=source.enabled,
+        )
+        for source in config.server.sources
+        if isinstance(source, PythonSourceConfig)
+    }
+
+
 def _build_mcp_workflow_server_from_workflow_config(
     config: WorkflowConfigFile,
 ) -> WorkflowServer:
@@ -49,7 +65,10 @@ def build_workflow_server_from_workflow_config(
         # Roadmap: SQL/transactional stores are deferred until the remote server
         # storage boundary is proven with file-backed stores.
         raise ValueError("wf-rpc-server currently requires filesystem store")
-    return build_local_static_workflow_server(store.root)
+    return build_local_static_workflow_server(
+        store.root,
+        extra_sources=_python_sources(config),
+    )
 
 
 def build_workflow_server_from_legacy_mcp_config(path: str | Path) -> WorkflowServer:

src/wf_server/context.py

@@ -268,12 +268,22 @@ class WorkflowServer:
         return TraceRange(start=start, limit=limit)
 
 
-def build_local_static_workflow_server(root: str | Path) -> WorkflowServer:
+def build_local_static_workflow_server(
+    root: str | Path,
+    *,
+    extra_sources: Mapping[str, CapabilitySource] | None = None,
+) -> WorkflowServer:
     """Build a durable local/static workflow server composition."""
     config = WorkflowServerConfig(store_root=Path(root))
     stores = file_workflow_stores(config.store_root)
     events = InMemoryWorkflowEventRecorder()
-    specs = StaticWorkflowSpecProvider(builtin_sources())
+    sources = builtin_sources()
+    if extra_sources:
+        overlap = set(sources) & set(extra_sources)
+        if overlap:
+            raise ValueError(f"duplicate workflow source ids: {sorted(overlap)}")
+        sources.update(extra_sources)
+    specs = StaticWorkflowSpecProvider(sources)
     runtime = LocalWorkflowRuntimeRunner(
         specs=specs,
         artifact_store=stores.artifact_store,

tests/wf_server/test_config_composition.py

@@ -12,6 +12,33 @@ from wf_server.config import (
 from wf_server.context import WorkflowServer
 
 
+def test_workflow_config_with_python_source_exposes_capability(tmp_path: Path) -> None:
+    config = WorkflowConfigFile.model_validate(
+        {
+            "version": 1,
+            "server": {
+                "store": {"kind": "filesystem", "root": str(tmp_path / "store")},
+                "sources": [
+                    {
+                        "kind": "python",
+                        "id": "local.ops",
+                        "module": "tests.fixtures.python_source_ops",
+                        "registry": "registry",
+                    }
+                ],
+            },
+        }
+    )
+
+    server = build_workflow_server_from_workflow_config(config)
+
+    assert "local.ops" in server.context.specs.capability_sources
+    assert (
+        "local.ops.echo"
+        in server.context.specs.capability_sources["local.ops"].capabilities.node_specs
+    )
+
+
 def test_build_workflow_server_from_workflow_config_uses_local_static_for_no_mcp_sources(
     tmp_path: Path,
 ) -> None: