Add entry-point discovery for tasks and workflows

README.md

@@ -22,25 +22,30 @@ pip install -e ".[dev]"
 from pydantic import BaseModel
 from taskmaestro import Task, Workflow, Job, Runner, ExecutionContext
 
+
 class NumberInput(BaseModel):
     value: int
 
+
 class NumberOutput(BaseModel):
     value: int
 
+
 class AddOne(Task[NumberInput, NumberOutput]):
     def run(self, input: NumberInput, ctx: ExecutionContext) -> NumberOutput:
         return NumberOutput(value=input.value + 1)
 
+
 class Double(Task[NumberOutput, NumberOutput]):
     def run(self, input: NumberOutput, ctx: ExecutionContext) -> NumberOutput:
         return NumberOutput(value=input.value * 2)
 
+
 workflow = Workflow(name="math", tasks=[AddOne, Double])
 job = Job(workflow=workflow, config=NumberInput(value=5))
 result = Runner().run(job)
 
-print(result.status)        # "completed"
+print(result.status)  # "completed"
 print(result.result.value)  # 12
 ```
 
@@ -50,31 +55,39 @@ print(result.result.value)  # 12
 from pydantic import BaseModel
 from taskmaestro import Task, Workflow, Job, Runner, ExecutionContext
 
+
 class Input(BaseModel):
     value: int
 
+
 class Output(BaseModel):
     value: int
 
+
 class MergedInput(BaseModel):
     a: Output
     b: Output
 
+
 class MergedOutput(BaseModel):
     total: int
 
+
 class BranchA(Task[Input, Output]):
     def run(self, input: Input, ctx: ExecutionContext) -> Output:
         return Output(value=input.value + 1)
 
+
 class BranchB(Task[Input, Output]):
     def run(self, input: Input, ctx: ExecutionContext) -> Output:
         return Output(value=input.value * 2)
 
+
 class Merge(Task[MergedInput, MergedOutput]):
     def run(self, input: MergedInput, ctx: ExecutionContext) -> MergedOutput:
         return MergedOutput(total=input.a.value + input.b.value)
 
+
 workflow = (
     Workflow.builder(name="fan_in")
     .add_task(BranchA)
@@ -86,7 +99,7 @@ workflow = (
 job = Job(workflow=workflow, config=Input(value=5))
 result = Runner().run(job)
 
-print(result.status)        # "completed"
+print(result.status)  # "completed"
 print(result.result.total)  # 16 (6 + 10)
 ```
 
@@ -131,15 +144,17 @@ from taskmaestro import EmptyConfig, Job, JobConfiguration, Workflow
 
 workflow = (
     Workflow.builder(name="configured")
-    .add_task(LoadModel, config_fields=["path"])          # root: all input from config
+    .add_task(LoadModel, config_fields=["path"])  # root: all input from config
     .add_task(Transform, depends_on=LoadModel, config_fields=["scale_factor"])  # mixed
     .build()
 )
 
-job_config = JobConfiguration({
-    "load_model": {"path": "/data/model.egrid"},
-    "transform": {"scale_factor": 2.5},
-})
+job_config = JobConfiguration(
+    {
+        "load_model": {"path": "/data/model.egrid"},
+        "transform": {"scale_factor": 2.5},
+    }
+)
 
 job = Job(workflow=workflow, config=EmptyConfig(), job_configuration=job_config)
 result = Runner().run(job)
@@ -160,15 +175,19 @@ class ExtractKeywords(Task):
 
     def run(self, input: Inputs, ctx: ExecutionContext) -> Outputs: ...
 
+
 workflow = (
     Workflow.builder(name="analysis")
     .add_task(ExtractKeywords, depends_on=PrepareText)
     .add_task(
         BuildReport,
         depends_on={
-            "keywords": (ExtractKeywords, "keywords"),              # routes .keywords field
-            "num_words_removed": (ExtractKeywords, "num_words_removed"),  # routes .num_words_removed
-            "stats": ComputeWordStats,                              # whole output
+            "keywords": (ExtractKeywords, "keywords"),  # routes .keywords field
+            "num_words_removed": (
+                ExtractKeywords,
+                "num_words_removed",
+            ),  # routes .num_words_removed
+            "stats": ComputeWordStats,  # whole output
         },
     )
     .build()
@@ -186,19 +205,51 @@ from taskmaestro import ObjectModel
 GridCase = ObjectModel[rips.EclipseCase]
 WellPath = ObjectModel[rips.WellPath]
 
+
 # Subclass — adds fields alongside the wrapped object
 class AddPerforationInput(ObjectModel[rips.WellPath]):
     start_md: float
     end_md: float
 
+
 # Access the wrapped object via .value
 grid = GridCase(value=eclipse_case)
 print(grid.value.name)
 ```
 
+## Plugin discovery
+
+Installed packages can publish tasks and workflows using standard Python entry points:
+
+```toml
+[project.entry-points."taskmaestro.tasks"]
+"acme.prepare" = "acme_tasks.prepare:Prepare"
+
+[project.entry-points."taskmaestro.workflows"]
+"acme.analysis" = "acme_tasks.workflows:analysis_workflow"
+```
+
+A task entry point must resolve to a `Task` subclass and a workflow entry point must
+resolve to a `Workflow` instance. Prefix names with the provider name to avoid clashes.
+Consumers can discover plugins without scanning package directories:
+
+```python
+from taskmaestro import registered_tasks, registered_workflows
+
+tasks = registered_tasks()  # dict[str, type[Task]]
+workflows = registered_workflows()  # dict[str, Workflow]
+```
+
+Use `registered_task_names()` and `registered_workflow_names()` to inspect identifiers
+without importing plugin modules, or `get_registered_task(name)` and
+`get_registered_workflow(name)` to load one plugin. Duplicate names and invalid plugin
+types raise `PluginLoadError`.
+
 ## YAML Configuration
 
-Workflows can be defined entirely in YAML instead of Python. The loader dynamically imports task classes and validates the full configuration:
+Workflows can be defined entirely in YAML instead of Python. A `task:` value may be
+either a registered task identifier or a dotted Python class path. The loader resolves
+registered identifiers first and validates the full configuration:
 
 ```yaml
 # workflow.yaml

examples/resinsight/pipeline.py

@@ -114,8 +114,12 @@ class AddPerforationInput(BaseModel):
 
     resinsight: RipsInstance
     well_path: WellPath
-    event_date: str = Field(description="Perforation event date in ISO format (YYYY-MM-DD)", examples=["2024-01-15"])
-    start_md: float = Field(description="Start measured depth of the perforation interval (m)", ge=0)
+    event_date: str = Field(
+        description="Perforation event date in ISO format (YYYY-MM-DD)", examples=["2024-01-15"]
+    )
+    start_md: float = Field(
+        description="Start measured depth of the perforation interval (m)", ge=0
+    )
     end_md: float = Field(description="End measured depth of the perforation interval (m)", ge=0)
 
 
@@ -221,14 +225,19 @@ class Inputs(BaseModel):
         grid_case: GridCase
         perforation_1: PerforationOutput
         perforation_2: PerforationOutput
-        event_date: str = Field(description="Timestamp written into the exported schedule file", examples=["2024-05-01"])
+        event_date: str = Field(
+            description="Timestamp written into the exported schedule file",
+            examples=["2024-05-01"],
+        )
         export_path: str = Field(description="Output path for the .sch completions file")
 
     class Outputs(BaseModel):
         """Exported completions summary."""
 
         export_file: str = Field(description="Path to the generated completions file")
-        well_path_names: list[str] = Field(description="Names of the well paths included in the export")
+        well_path_names: list[str] = Field(
+            description="Names of the well paths included in the export"
+        )
 
     def run(self, input: Inputs, ctx: ExecutionContext) -> Outputs:
         eclipse_case = input.grid_case.value

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "taskmaestro"
-version = "0.1.0"
+version = "0.2.0"
 description = "A Python 3.12+ library for defining and executing typed DAG task workflows"
 readme = "README.md"
 requires-python = ">=3.12"
@@ -44,7 +44,7 @@ packages = ["taskmaestro"]
 dev = [
     "pytest>=8.0",
     "pytest-cov>=5.0",
-    "ruff>=0.4",
+    "ruff>=0.16",
     "mypy>=1.10",
     "types-PyYAML>=6.0",
 ]

taskmaestro/__init__.py

@@ -1,13 +1,24 @@
 """Workflow Runner: typed DAG task workflows with Pydantic models."""
 
-__version__ = "0.1.0"
+__version__ = "0.2.0"
 
 from taskmaestro.context import ExecutionContext
+from taskmaestro.discovery import (
+    TASK_ENTRY_POINT_GROUP,
+    WORKFLOW_ENTRY_POINT_GROUP,
+    get_registered_task,
+    get_registered_workflow,
+    registered_task_names,
+    registered_tasks,
+    registered_workflow_names,
+    registered_workflows,
+)
 from taskmaestro.exceptions import (
     ConfigLoadError,
     CycleDetectedError,
     IncompleteInputError,
     JobStateError,
+    PluginLoadError,
     TaskExecutionError,
     TaskOutputTypeError,
     TaskTimeoutError,
@@ -28,6 +39,8 @@
 )
 
 __all__ = [
+    "TASK_ENTRY_POINT_GROUP",
+    "WORKFLOW_ENTRY_POINT_GROUP",
     "ConfigLoadError",
     "CycleDetectedError",
     "EmptyConfig",
@@ -39,6 +52,7 @@
     "JobStatus",
     "LoadedWorkflow",
     "ObjectModel",
+    "PluginLoadError",
     "Runner",
     "Task",
     "TaskExecutionError",
@@ -50,7 +64,13 @@
     "WorkflowBuilder",
     "WorkflowDefinitionError",
     "WorkflowRunnerError",
+    "get_registered_task",
+    "get_registered_workflow",
     "load_workflow_from_yaml",
+    "registered_task_names",
+    "registered_tasks",
+    "registered_workflow_names",
+    "registered_workflows",
     "run_workflow_from_yaml",
     "to_mermaid",
     "workflow_task",

taskmaestro/discovery.py

@@ -0,0 +1,96 @@
+"""Discovery of tasks and workflows published by installed distributions."""
+
+from __future__ import annotations
+
+from importlib.metadata import EntryPoint, entry_points
+from typing import Any
+
+from taskmaestro.exceptions import PluginLoadError
+from taskmaestro.task import Task
+from taskmaestro.workflow import Workflow
+
+TASK_ENTRY_POINT_GROUP = "taskmaestro.tasks"
+WORKFLOW_ENTRY_POINT_GROUP = "taskmaestro.workflows"
+
+
+def _entry_points_by_name(group: str) -> dict[str, EntryPoint]:
+    """Return entry points in *group*, rejecting ambiguous registrations."""
+    result: dict[str, EntryPoint] = {}
+    for entry_point in entry_points(group=group):
+        if entry_point.name in result:
+            raise PluginLoadError(
+                f"Multiple entry points named '{entry_point.name}' are registered in '{group}'"
+            )
+        result[entry_point.name] = entry_point
+    return result
+
+
+def _load[T](entry_point: EntryPoint, expected_type: type[T], kind: str) -> T:
+    try:
+        value: Any = entry_point.load()
+    except Exception as exc:
+        raise PluginLoadError(
+            f"Cannot load {kind} entry point '{entry_point.name}' ({entry_point.value}): {exc}"
+        ) from exc
+    if not isinstance(value, expected_type):
+        raise PluginLoadError(
+            f"{kind.capitalize()} entry point '{entry_point.name}' ({entry_point.value}) "
+            f"must resolve to a {expected_type.__name__}"
+        )
+    return value
+
+
+def registered_task_names() -> set[str]:
+    """Return registered task identifiers without importing their packages."""
+    return set(_entry_points_by_name(TASK_ENTRY_POINT_GROUP))
+
+
+def registered_workflow_names() -> set[str]:
+    """Return registered workflow identifiers without importing their packages."""
+    return set(_entry_points_by_name(WORKFLOW_ENTRY_POINT_GROUP))
+
+
+def registered_tasks() -> dict[str, type[Task[Any, Any]]]:
+    """Load all tasks registered in the ``taskmaestro.tasks`` entry-point group.
+
+    The returned mapping is keyed by the entry-point name, which is the stable
+    identifier that configuration files and discovery clients should use.
+    """
+    tasks: dict[str, type[Task[Any, Any]]] = {}
+    for name, entry_point in _entry_points_by_name(TASK_ENTRY_POINT_GROUP).items():
+        task = _load(entry_point, type, "task")
+        if not issubclass(task, Task):
+            raise PluginLoadError(
+                f"Task entry point '{name}' ({entry_point.value}) must resolve to a Task subclass"
+            )
+        tasks[name] = task
+    return tasks
+
+
+def registered_workflows() -> dict[str, Workflow]:
+    """Load all workflows registered in the ``taskmaestro.workflows`` group."""
+    return {
+        name: _load(entry_point, Workflow, "workflow")
+        for name, entry_point in _entry_points_by_name(WORKFLOW_ENTRY_POINT_GROUP).items()
+    }
+
+
+def get_registered_task(name: str) -> type[Task[Any, Any]]:
+    """Load one registered task by its entry-point name."""
+    entry_point = _entry_points_by_name(TASK_ENTRY_POINT_GROUP).get(name)
+    if entry_point is None:
+        raise PluginLoadError(f"No task entry point named '{name}' is registered")
+    task = _load(entry_point, type, "task")
+    if not issubclass(task, Task):
+        raise PluginLoadError(
+            f"Task entry point '{name}' ({entry_point.value}) must resolve to a Task subclass"
+        )
+    return task
+
+
+def get_registered_workflow(name: str) -> Workflow:
+    """Load one registered workflow by its entry-point name."""
+    entry_point = _entry_points_by_name(WORKFLOW_ENTRY_POINT_GROUP).get(name)
+    if entry_point is None:
+        raise PluginLoadError(f"No workflow entry point named '{name}' is registered")
+    return _load(entry_point, Workflow, "workflow")

taskmaestro/exceptions.py

@@ -35,3 +35,7 @@ class TaskTimeoutError(TaskExecutionError):
 
 class ConfigLoadError(WorkflowRunnerError):
     """Raised when YAML config loading fails (parse errors, import failures, validation)."""
+
+
+class PluginLoadError(WorkflowRunnerError):
+    """Raised when an installed task or workflow entry point is invalid."""