lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

the WorkflowApiBackend collapses

Browse Source
This commit is contained in:
lda
2026-06-02 11:48:55 +07:00 Verified
parent 09995d86d5
commit 55d45226f4
19 changed files with 984 additions and 1385 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+6 -6
View File
@@ -40,9 +40,9 @@ implementation state.
6. **Workflow API seam**
- `wf_api.WorkflowApi` is now the process-local application-facing workflow
API used by both CLI commands and MCP workflow tools.
- `wf_api` imports no `wf_mcp` modules. The current adapter,
`WfMcpWorkflowApiBackend`, still wraps the existing MCP service stack while
later slices extract protocol-neutral logic behind that seam.
- `wf_api` imports no `wf_mcp` modules. `WorkflowApi` composes domain
services directly from `WorkflowOperationContext`; MCP owns only context
construction and tool schemas.
- Protocol-neutral helpers now live in `wf_api`: refs/constants, wrapper
hints, next actions, raw workflow plans, runtime dependencies, saved
subgraph preparation, and durable run lifecycle helpers. Old
@@ -115,9 +115,9 @@ implementation state.
Protocol-neutral operation context and domain services now exist behind
`wf_api`; MCP tool schemas and tool registration stay in `wf_mcp`.
- Workflow store ownership is explicit: entrypoints construct/inject `WorkflowStores`; `WfMcpService` no longer guesses stores from the MCP store root.
- The next useful slice is removing the remaining double-delegation path so
`WorkflowApi` composes domain services directly instead of routing through
`WfMcpWorkflowApiBackend` and `WorkflowSurfaceHandlers`.
- Double-delegation has been removed: CLI and MCP workflow tools construct
`WorkflowApi(context_from_service(service))` directly. `WorkflowSurfaceHandlers`
remains only as a temporary compatibility shim for older imports.
Frame stress points remaining for native subgraphs and future fork/gather:
docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md
+5
View File
@@ -4,6 +4,11 @@
**Goal:** Extract a protocol-neutral workflow application API from `wf_mcp` while preserving current process-local behavior and avoiding a large semantic rewrite.
> Current update: the original `WorkflowApiBackend` seam was useful for proving
> dependency direction, but has been collapsed. `WorkflowApi` now composes
> domain services directly from `WorkflowOperationContext`; MCP owns only
> context construction and tool schemas.
**Architecture:** `wf_api` becomes the long-lived in-process application service layer. `wf_mcp`, `wf_cli`, and future HTTP/UI adapters call `wf_api`; `wf_api` must not import `wf_mcp`.
**Current State:** Slice 1 introduced `wf_api.WorkflowApi`,
docs/superpowers/plans/2026-06-02-wf-api-remove-double-delegation.md
+754
View File
@@ -0,0 +1,754 @@
# wf_api Remove Double Delegation Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Collapse workflow calls from `WorkflowApi -> WfMcpWorkflowApiBackend -> WorkflowSurfaceHandlers -> domain services` to `WorkflowApi -> domain services`.
**Architecture:** `WorkflowApi` becomes the protocol-neutral application facade that composes `WorkflowCapabilityApi`, `WorkflowDraftApi`, `WorkflowArtifactApi`, `WorkflowDeploymentApi`, and `WorkflowRunApi` from a `WorkflowOperationContext`. MCP and CLI construct `WorkflowApi(context_from_service(service))` directly. `WorkflowSurfaceHandlers` remains only as a temporary compatibility shim for legacy imports/tests, and `WfMcpWorkflowApiBackend` / `WorkflowApiBackend` are removed.
**Tech Stack:** Python 3.14, `wf_api`, `wf_mcp`, dataclasses, pytest, ruff, basedpyright.
---
## Current Chain
```text
wf_mcp tools / wf_cli
-> WorkflowApi
-> WfMcpWorkflowApiBackend
-> WorkflowSurfaceHandlers
-> WorkflowCapabilityApi / WorkflowDraftApi / WorkflowArtifactApi / WorkflowDeploymentApi / WorkflowRunApi
```
This creates two mechanical delegation layers. Adding one workflow operation currently requires touching at least `WorkflowApi`, `WorkflowApiBackend`, `WfMcpWorkflowApiBackend`, and usually `WorkflowSurfaceHandlers`.
## Target Chain
```text
wf_mcp tools / wf_cli
-> WorkflowApi
-> WorkflowCapabilityApi / WorkflowDraftApi / WorkflowArtifactApi / WorkflowDeploymentApi / WorkflowRunApi
```
Legacy imports of `WorkflowSurfaceHandlers` may still work, but only as a thin wrapper around `WorkflowApi`.
## Files
- Modify: `src/wf_api/models.py`
- Modify: `src/wf_api/__init__.py`
- Modify: `src/wf_api/service.py`
- Delete: `src/wf_api/backend.py`
- Delete: `src/wf_mcp/broker/service/workflow_api_backend.py`
- Modify: `src/wf_cli/context.py`
- Modify: `src/wf_cli/commands/runs.py`
- Modify: `src/wf_mcp/workflow_surface/tools.py`
- Modify: `src/wf_mcp/workflow_surface/handlers.py`
- Modify: `tests/wf_cli/test_context.py`
- Modify: `tests/wf_api/test_cli_context_uses_api.py`
- Add: `tests/wf_api/test_direct_service.py`
- Add: `tests/wf_api/test_no_double_delegation.py`
- Modify docs: `docs/current_roadmap.md`, `docs/wf_mcp_architecture.md`, `docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md`
---
### Task 1: Move TraceRange Out of `backend.py`
**Files:**
- Modify: `src/wf_api/models.py`
- Modify: `src/wf_api/__init__.py`
- Modify: `src/wf_cli/commands/runs.py`
- Modify: `src/wf_mcp/workflow_surface/tools.py`
- [ ] **Step 1: Write import regression test**
Add to `tests/wf_api/test_raw_workflow_plan_extraction.py`:
```python
def test_trace_range_exports_from_wf_api_models() -> None:
from wf_api import TraceRange
from wf_api.models import TraceRange as CanonicalTraceRange
assert TraceRange is CanonicalTraceRange
assert TraceRange(start=1, limit=2).start == 1
assert TraceRange(start=1, limit=2).limit == 2
```
- [ ] **Step 2: Run failing test**
Run:
```bash
uv run pytest tests\wf_api\test_raw_workflow_plan_extraction.py::test_trace_range_exports_from_wf_api_models -q
```
Expected: fail because `wf_api.models.TraceRange` does not exist yet.
- [ ] **Step 3: Add `TraceRange` to `wf_api.models`**
In `src/wf_api/models.py`, add imports:
```python
from dataclasses import dataclass
```
Then add before `RawWorkflowPlan`:
```python
@dataclass(frozen=True, slots=True)
class TraceRange:
"""Caller-bounded debug trace slice for durable deployment runs."""
start: int = 0
limit: int = 25
```
- [ ] **Step 4: Update `wf_api.__init__` export**
Change:
```python
from .backend import TraceRange, WorkflowApiBackend
```
to:
```python
from .models import RawWorkflowPlan, TraceRange
```
Remove `"WorkflowApiBackend"` from `__all__`. Keep `"TraceRange"`.
If `RawWorkflowPlan` was not exported before, include it only if already expected by tests; do not add new public API unless the existing file already imports it elsewhere.
- [ ] **Step 5: Update imports that referenced `wf_api.backend.TraceRange`**
In `src/wf_cli/commands/runs.py`, replace:
```python
from wf_api.backend import TraceRange
```
with:
```python
from wf_api import TraceRange
```
In `src/wf_mcp/workflow_surface/tools.py`, remove:
```python
from wf_api.backend import TraceRange as ApiTraceRange
```
Also remove `_to_api_trace_range()`. Later tasks pass MCP `TraceRange` directly because `WorkflowRunApi` validates trace ranges structurally through `TraceRangeLike`.
- [ ] **Step 6: Verify Task 1**
Run:
```bash
uv run pytest tests\wf_api\test_raw_workflow_plan_extraction.py::test_trace_range_exports_from_wf_api_models tests\wf_cli\test_run_deploy.py tests\wf_mcp\server\test_tools.py -q
uv run ruff check src\wf_api\models.py src\wf_api\__init__.py src\wf_cli\commands\runs.py src\wf_mcp\workflow_surface\tools.py tests\wf_api\test_raw_workflow_plan_extraction.py
uv run ruff format --check src\wf_api\models.py src\wf_api\__init__.py src\wf_cli\commands\runs.py src\wf_mcp\workflow_surface\tools.py tests\wf_api\test_raw_workflow_plan_extraction.py
```
Expected: tests pass, lint pass, format pass.
---
### Task 2: Make `WorkflowApi` Compose Domain Services Directly
**Files:**
- Modify: `src/wf_api/service.py`
- Add: `tests/wf_api/test_direct_service.py`
- [ ] **Step 1: Write direct-composition tests**
Create `tests/wf_api/test_direct_service.py`:
```python
from __future__ import annotations
import asyncio
from wf_artifacts import FileWorkflowArtifactStore
from wf_api import WorkflowApi
from wf_api.artifacts import WorkflowArtifactApi
from wf_api.capabilities import WorkflowCapabilityApi
from wf_api.deployments import WorkflowDeploymentApi
from wf_api.drafts import WorkflowDraftApi
from wf_api.runs import WorkflowRunApi
from wf_mcp.broker import WfMcpService
from wf_mcp.broker.service.workflow_operation_context import context_from_service
from wf_mcp.models import ConnectionConfig
from wf_mcp.storage import FileStore
from tests.wf_mcp.test_support import echo_tool, local_temp_root
def _api() -> WorkflowApi:
root = local_temp_root() / "wf_api_direct_composition"
service = WfMcpService(
store=FileStore(root / "mcp"),
artifact_store=FileWorkflowArtifactStore(root),
)
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool)
return WorkflowApi(context_from_service(service))
def test_workflow_api_composes_domain_services() -> None:
api = _api()
assert isinstance(api.capabilities, WorkflowCapabilityApi)
assert isinstance(api.drafts, WorkflowDraftApi)
assert isinstance(api.artifacts, WorkflowArtifactApi)
assert isinstance(api.deployments, WorkflowDeploymentApi)
assert isinstance(api.runs, WorkflowRunApi)
assert not hasattr(api, "backend")
def test_workflow_api_direct_capability_call() -> None:
api = _api()
result = asyncio.run(
api.call_capability(
qualified_name="demo.personal.echo_tool",
payload={"text": "hello"},
)
)
assert result["kind"] == "node_spec"
assert result["outcome"] == "ok"
assert result["output"] == {"echoed": "hello"}
```
- [ ] **Step 2: Run failing tests**
Run:
```bash
uv run pytest tests\wf_api\test_direct_service.py -q
```
Expected: fail because `WorkflowApi` still expects a `WorkflowApiBackend`.
- [ ] **Step 3: Rewrite `WorkflowApi.__init__`**
In `src/wf_api/service.py`, replace:
```python
from .backend import TraceRange, WorkflowApiBackend
```
with:
```python
from .artifacts import WorkflowArtifactApi
from .capabilities import WorkflowCapabilityApi
from .deployments import WorkflowDeploymentApi
from .drafts import WorkflowDraftApi
from .models import TraceRange
from .operation_context import WorkflowOperationContext
from .runs import TraceRangeLike, WorkflowRunApi
```
Replace the class docstring and constructor:
```python
class WorkflowApi:
"""Protocol-neutral workflow application facade.
This facade owns the stable application entry point. It composes the
domain APIs from a WorkflowOperationContext so MCP, CLI, and future HTTP
callers share one operation surface without importing wf_mcp.
"""
def __init__(self, context: WorkflowOperationContext) -> None:
self.context = context
self.capabilities = WorkflowCapabilityApi(context)
self.drafts = WorkflowDraftApi(context)
self.artifacts = WorkflowArtifactApi(context)
self.deployments = WorkflowDeploymentApi(context)
self.runs = WorkflowRunApi(context)
```
- [ ] **Step 4: Replace backend delegations with domain service delegations**
In `src/wf_api/service.py`, replace these groups:
Capabilities:
```python
self.backend.list_capabilities(...) -> self.capabilities.list_capabilities(...)
self.backend.inspect_capability(...) -> self.capabilities.inspect_capability(...)
self.backend.call_capability(...) -> self.capabilities.call_capability(...)
self.backend.create_draft_workspace_from_capability(...) -> self.capabilities.create_draft_workspace_from_capability(...)
```
Artifacts:
```python
self.backend.list_artifacts(...) -> self.artifacts.list_artifacts(...)
self.backend.inspect_artifact(...) -> self.artifacts.inspect_artifact(...)
self.backend.save_artifact(...) -> self.artifacts.save_artifact(...)
self.backend.create_artifact_from_plan(...) -> self.artifacts.create_artifact_from_plan(...)
self.backend.create_artifact_from_draft(...) -> self.artifacts.create_artifact_from_draft(...)
self.backend.create_artifact_from_workspace(...) -> self.artifacts.create_artifact_from_workspace(...)
self.backend.create_wrapper_from_workspace(...) -> self.artifacts.create_wrapper_from_workspace(...)
```
Drafts:
```python
self.backend.validate_draft(...) -> self.drafts.validate_draft(...)
self.backend.compile_draft(...) -> self.drafts.compile_draft(...)
self.backend.patch_draft(...) -> self.drafts.patch_draft(...)
self.backend.list_draft_workspaces() -> self.drafts.list_draft_workspaces()
self.backend.create_draft_workspace(...) -> self.drafts.create_draft_workspace(...)
self.backend.get_draft_workspace(...) -> self.drafts.get_draft_workspace(...)
self.backend.delete_draft_workspace(...) -> self.drafts.delete_draft_workspace(...)
self.backend.validate_draft_workspace(...) -> self.drafts.validate_draft_workspace(...)
self.backend.patch_draft_workspace(...) -> self.drafts.patch_draft_workspace(...)
self.backend.set_draft_name(...) -> self.drafts.set_draft_name(...)
self.backend.set_draft_route(...) -> self.drafts.set_draft_route(...)
self.backend.set_step_input_map(...) -> self.drafts.set_step_input_map(...)
self.backend.set_step_output_map(...) -> self.drafts.set_step_output_map(...)
self.backend.create_minimal_draft_workspace(...) -> self.drafts.create_minimal_draft_workspace(...)
```
Deployments:
```python
self.backend.list_deployments() -> self.deployments.list_deployments()
self.backend.inspect_deployment(...) -> self.deployments.inspect_deployment(...)
self.backend.save_deployment(...) -> self.deployments.save_deployment(...)
self.backend.delete_deployment(...) -> self.deployments.delete_deployment(...)
self.backend.validate_deployment(...) -> self.deployments.validate_deployment(...)
```
Runs:
```python
self.backend.run_deployment(...) -> self.runs.run_deployment(...)
self.backend.resume_run(...) -> self.runs.resume_run(...)
self.backend.inspect_run(...) -> self.runs.inspect_run(...)
self.backend.read_run_trace(...) -> self.runs.read_run_trace(...)
```
For run methods, change type hints from `TraceRange | None` to `TraceRangeLike | None` and `TraceRange` to `TraceRangeLike` so MCP Pydantic `TraceRange` and CLI dataclass `TraceRange` both remain accepted structurally.
- [ ] **Step 5: Verify Task 2**
Run:
```bash
uv run pytest tests\wf_api\test_direct_service.py tests\wf_api\test_capability_api.py tests\wf_api\test_drafts_service.py tests\wf_api\test_artifact_api.py tests\wf_api\test_deployment_api.py tests\wf_api\test_run_api.py -q
uv run ruff check src\wf_api\service.py tests\wf_api\test_direct_service.py
uv run ruff format --check src\wf_api\service.py tests\wf_api\test_direct_service.py
```
Expected: direct service tests and domain API tests pass.
---
### Task 3: Update CLI and MCP Tool Construction
**Files:**
- Modify: `src/wf_cli/context.py`
- Modify: `tests/wf_cli/test_context.py`
- Modify: `src/wf_mcp/workflow_surface/tools.py`
- Add: `tests/wf_api/test_no_double_delegation.py`
- [ ] **Step 1: Write no-backend-chain test**
Create `tests/wf_api/test_no_double_delegation.py`:
```python
from __future__ import annotations
import ast
from pathlib import Path
def _imports_module(path: Path, module_name: str) -> bool:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module == module_name:
return True
if isinstance(node, ast.Import):
if any(alias.name == module_name for alias in node.names):
return True
return False
def test_cli_and_mcp_tools_do_not_import_backend_adapter() -> None:
root = Path(__file__).resolve().parents[2]
assert not _imports_module(
root / "src" / "wf_cli" / "context.py",
"wf_mcp.broker.service.workflow_api_backend",
)
assert not _imports_module(
root / "src" / "wf_mcp" / "workflow_surface" / "tools.py",
"wf_mcp.broker.service.workflow_api_backend",
)
```
- [ ] **Step 2: Run failing no-backend-chain test**
Run:
```bash
uv run pytest tests\wf_api\test_no_double_delegation.py -q
```
Expected: fail because CLI context and MCP tools still import `WfMcpWorkflowApiBackend`.
- [ ] **Step 3: Update CLI context**
In `src/wf_cli/context.py`, remove:
```python
from wf_mcp.broker.service.workflow_api_backend import WfMcpWorkflowApiBackend
```
Add:
```python
from wf_mcp.broker.service.workflow_operation_context import context_from_service
```
Change `load_cli_context()`:
```python
handlers=WorkflowApi(context_from_service(service)),
```
- [ ] **Step 4: Update CLI context test**
In `tests/wf_cli/test_context.py`, replace the private backend-chain assertion:
```python
assert context.handlers.backend._handlers.service is context.service # type: ignore[attr-defined]
```
with:
```python
assert context.handlers.context.artifact_store is context.service.artifact_store
assert context.handlers.context.draft_workspace_store is context.service.draft_workspace_store
assert context.handlers.context.run_store is context.service.run_store
```
This tests the public context seam instead of the deleted backend chain.
- [ ] **Step 5: Update MCP workflow tools**
In `src/wf_mcp/workflow_surface/tools.py`, remove:
```python
from wf_mcp.broker.service.workflow_api_backend import WfMcpWorkflowApiBackend
```
Add:
```python
from wf_mcp.broker.service.workflow_operation_context import context_from_service
```
Change:
```python
handlers = WorkflowApi(WfMcpWorkflowApiBackend(service))
```
to:
```python
handlers = WorkflowApi(context_from_service(service))
```
For run tools, pass `request.trace_range` or `trace_range` directly to `handlers.*`. Remove conversions through `ApiTraceRange`.
- [ ] **Step 6: Verify Task 3**
Run:
```bash
uv run pytest tests\wf_api\test_no_double_delegation.py tests\wf_cli\test_context.py tests\wf_cli tests\wf_mcp\server\test_tools.py tests\wf_mcp\workflow_surface -q
uv run ruff check src\wf_cli\context.py src\wf_mcp\workflow_surface\tools.py tests\wf_cli\test_context.py tests\wf_api\test_no_double_delegation.py
uv run ruff format --check src\wf_cli\context.py src\wf_mcp\workflow_surface\tools.py tests\wf_cli\test_context.py tests\wf_api\test_no_double_delegation.py
```
Expected: CLI and MCP workflow tool tests pass.
---
### Task 4: Shrink `WorkflowSurfaceHandlers` to Compatibility Shim
**Files:**
- Modify: `src/wf_mcp/workflow_surface/handlers.py`
- Modify: `tests/wf_api/test_direct_service.py` or add a small handler shim test
- [ ] **Step 1: Add compatibility shim test**
Add to `tests/wf_api/test_direct_service.py`:
```python
def test_workflow_surface_handlers_is_compatibility_shim() -> None:
from wf_api import WorkflowApi
from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
root = local_temp_root() / "workflow_surface_handler_shim"
service = WfMcpService(
store=FileStore(root / "mcp"),
artifact_store=FileWorkflowArtifactStore(root),
)
handlers = WorkflowSurfaceHandlers(service)
assert isinstance(handlers, WorkflowApi)
assert handlers.service is service
assert handlers.context.artifact_store is service.artifact_store
```
- [ ] **Step 2: Run failing compatibility test**
Run:
```bash
uv run pytest tests\wf_api\test_direct_service.py::test_workflow_surface_handlers_is_compatibility_shim -q
```
Expected: fail because `WorkflowSurfaceHandlers` is not a `WorkflowApi` subclass yet.
- [ ] **Step 3: Replace `WorkflowSurfaceHandlers` implementation**
Replace `src/wf_mcp/workflow_surface/handlers.py` with:
```python
from __future__ import annotations
from wf_api import WorkflowApi
from ..broker.service.workflow_operation_context import context_from_service
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..broker.service import WfMcpService
class WorkflowSurfaceHandlers(WorkflowApi):
"""Compatibility wrapper for old wf_mcp.workflow_surface imports.
New code should construct `WorkflowApi(context_from_service(service))`
directly. This shim keeps tests and legacy broker artifact tools working
while the MCP surface is migrated.
"""
def __init__(self, service: WfMcpService) -> None:
self.service = service
super().__init__(context_from_service(service))
__all__ = ["WorkflowSurfaceHandlers"]
```
This file should no longer import domain services directly.
- [ ] **Step 4: Verify handler compatibility**
Run:
```bash
uv run pytest tests\wf_api\test_direct_service.py::test_workflow_surface_handlers_is_compatibility_shim tests\wf_mcp\workflow_surface tests\wf_mcp\test_saved_subgraphs.py tests\wf_mcp\broker -q
uv run ruff check src\wf_mcp\workflow_surface\handlers.py tests\wf_api\test_direct_service.py
uv run ruff format --check src\wf_mcp\workflow_surface\handlers.py tests\wf_api\test_direct_service.py
```
Expected: old handler tests pass through the shim.
---
### Task 5: Delete Backend Protocol and Adapter
**Files:**
- Delete: `src/wf_api/backend.py`
- Delete: `src/wf_mcp/broker/service/workflow_api_backend.py`
- Modify: `src/wf_api/__init__.py`
- Modify docs that describe the old backend chain
- [ ] **Step 1: Delete backend files**
Delete:
```text
src/wf_api/backend.py
src/wf_mcp/broker/service/workflow_api_backend.py
```
- [ ] **Step 2: Remove public backend export**
In `src/wf_api/__init__.py`, ensure there is no import or `__all__` entry for `WorkflowApiBackend`.
- [ ] **Step 3: Search for live backend references**
Run:
```bash
rg -n "WorkflowApiBackend|WfMcpWorkflowApiBackend|workflow_api_backend|\\.backend" src tests
```
Expected: no live source/test references.
Historical docs under `docs/superpowers/plans/2026-06-01-*` may still mention the old slice. Do not rewrite historical plans except the active roadmap files named in Task 6.
- [ ] **Step 4: Verify deletion**
Run:
```bash
uv run pytest tests\wf_api\test_import_direction.py tests\wf_api\test_no_double_delegation.py tests\wf_api\test_cli_context_uses_api.py -q
uv run ruff check src\wf_api src\wf_cli\context.py src\wf_mcp\workflow_surface src\wf_mcp\broker\service tests\wf_api
uv run ruff format --check src\wf_api src\wf_cli\context.py src\wf_mcp\workflow_surface src\wf_mcp\broker\service tests\wf_api
```
Expected: tests pass, lint pass, format pass.
---
### Task 6: Update Active Docs
**Files:**
- Modify: `docs/current_roadmap.md`
- Modify: `docs/wf_mcp_architecture.md`
- Modify: `docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md`
- [ ] **Step 1: Update `docs/current_roadmap.md`**
Replace the bullet that says the next useful slice is removing double-delegation with:
```markdown
- Double-delegation has been removed: CLI and MCP workflow tools construct
`WorkflowApi(context_from_service(service))` directly. `WorkflowSurfaceHandlers`
remains only as a temporary compatibility shim for older imports.
```
- [ ] **Step 2: Update `docs/wf_mcp_architecture.md`**
Find the architecture text that contains:
```text
wf_api.WorkflowApi ───> WorkflowApiBackend
```
Replace that diagram/text with:
```text
wf_mcp.workflow_surface.tools
-> wf_api.WorkflowApi
-> wf_api domain services
-> WorkflowOperationContext
-> WfMcpService adapters/stores/runtime
```
Add:
```markdown
`WorkflowSurfaceHandlers` is a compatibility shim only. New entrypoints should
construct `WorkflowApi(context_from_service(service))` directly.
```
- [ ] **Step 3: Update active extraction roadmap**
In `docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md`, add a current-state note near the top:
```markdown
> Current update: the original `WorkflowApiBackend` seam was useful for proving
> dependency direction, but has been collapsed. `WorkflowApi` now composes
> domain services directly from `WorkflowOperationContext`; MCP owns only
> context construction and tool schemas.
```
Do not rewrite the historical task bodies. They describe prior slices.
- [ ] **Step 4: Verify docs**
Run:
```bash
git diff --check -- docs\current_roadmap.md docs\wf_mcp_architecture.md docs\superpowers\plans\2026-06-01-wf-api-extraction-roadmap.md
```
Expected: no whitespace errors.
---
### Task 7: Final Verification
**Files:**
- All touched files.
- [ ] **Step 1: Run focused workflow API/MCP/CLI tests**
Run:
```bash
uv run pytest tests\wf_api tests\wf_cli tests\wf_mcp\workflow_surface tests\wf_mcp\server\test_tools.py tests\wf_mcp\test_saved_subgraphs.py -q
```
Expected: selected tests pass.
- [ ] **Step 2: Run full suite**
Run:
```bash
uv run pytest -q
```
Expected: full suite passes with known skip/xfail counts.
- [ ] **Step 3: Run lint and format checks**
Run:
```bash
uv run ruff check src\wf_api src\wf_cli src\wf_mcp tests\wf_api tests\wf_cli tests\wf_mcp
uv run ruff format --check src\wf_api src\wf_cli src\wf_mcp tests\wf_api tests\wf_cli tests\wf_mcp
```
Expected: all checks pass.
- [ ] **Step 4: Run typecheck**
Run:
```bash
uv run basedpyright --level error
```
Expected: `0 errors, 0 warnings, 0 notes`. If the command exits nonzero only because of the known workspace enumeration warning, report that exactly.
- [ ] **Step 5: Final reference check**
Run:
```bash
rg -n "WorkflowApiBackend|WfMcpWorkflowApiBackend|workflow_api_backend|\\.backend" src tests docs\current_roadmap.md docs\wf_mcp_architecture.md
```
Expected: no references in live source/tests/current docs.
---
## Self-Review
- Spec coverage: The plan removes `WorkflowApiBackend`, deletes `WfMcpWorkflowApiBackend`, updates CLI/MCP construction, keeps handler compatibility, moves `TraceRange`, and updates active docs.
- Placeholder scan: No `TODO`/`TBD` placeholders remain.
- Type consistency: `WorkflowApi` accepts `WorkflowOperationContext`; run trace methods accept `TraceRangeLike`; `TraceRange` is a convenience DTO exported from `wf_api.models`.
- Scope check: This does not remove all `WorkflowSurfaceHandlers` tests or legacy imports. It reduces handlers to a shim; deleting the shim is a later cleanup once `wf_mcp.broker.artifact_tools` and legacy tests stop importing it.
docs/wf_mcp_architecture.md
+13 -20
View File
@@ -41,19 +41,24 @@ relevant concern package directly.
Workflow lifecycle operations now have a protocol-neutral front door:
```text
wf_cli ─┐
├──> wf_api.WorkflowApi ───> WorkflowApiBackend
wf_mcp ─┘
wf_mcp.workflow_surface.tools
-> wf_api.WorkflowApi
-> wf_api domain services
-> WorkflowOperationContext
-> WfMcpService adapters/stores/runtime
```
The current backend is `wf_mcp.broker.service.WfMcpWorkflowApiBackend`, which
wraps the existing `wf_mcp.workflow_surface.WorkflowSurfaceHandlers`. That
handler class still contains most workflow-surface logic and still depends on
`WfMcpService`; it is kept for compatibility and incremental extraction.
`WorkflowSurfaceHandlers` is a compatibility shim only. New entrypoints should
construct `WorkflowApi(context_from_service(service))` directly.
The old backend-adapter/protocol layer has been removed. `WorkflowApi` composes
domain services (`WorkflowCapabilityApi`, `WorkflowDraftApi`,
`WorkflowArtifactApi`, `WorkflowDeploymentApi`, `WorkflowRunApi`) from a
`WorkflowOperationContext`.
New code should treat `wf_api.WorkflowApi` as the application-facing API. Do not
add new callers that import `WorkflowSurfaceHandlers` directly unless they are
inside the MCP backend adapter or compatibility tests.
compatibility tests.
This is a dependency-direction cleanup, not a full domain split. Most API
methods still mirror the old workflow-surface payloads and return
@@ -66,18 +71,6 @@ old `wf_mcp.workflow_surface.*` helper modules are compatibility shims. MCP
tool schemas and registration still live in `wf_mcp.workflow_surface.models` and
`wf_mcp.workflow_surface.tools`.
The remaining extraction work is the large operation implementation:
`WorkflowSurfaceHandlers` still owns most capability, draft, artifact,
deployment, and run methods. Split that implementation by domain only after the
current helper shims are stable.
Do not move those method bodies directly into `wf_api` while they still depend
on the whole `WfMcpService`. The next extraction step should define a
protocol-neutral operation context/protocol seam for stores, capability sources,
event recording, and live source calls. MCP-owned code can adapt `WfMcpService`
into that seam; domain services can then depend on the seam rather than on MCP
service internals.
## Broker Catalogs
The broker keeps two related catalog views:
src/wf_api/__init__.py
+2 -2
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from .listing import matches_query, paged_list_payload
from .artifacts import WorkflowArtifactApi
from .backend import TraceRange, WorkflowApiBackend
from .models import RawWorkflowPlan, TraceRange
from .capabilities import WorkflowCapabilityApi
from .constants import (
DEFAULT_CALL_STEP_ID,
@@ -56,10 +56,10 @@ __all__ = [
"OutcomeCandidate",
"OutcomeCandidateKind",
"RUNTIME_ERROR_CAPABILITY",
"RawWorkflowPlan",
"RuntimeDependencies",
"TraceRange",
"WorkflowApi",
"WorkflowApiBackend",
"WorkflowArtifactApi",
"WorkflowCapabilityApi",
"WorkflowArtifactCataloger",
src/wf_api/backend.py
-321
View File
@@ -1,321 +0,0 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, Protocol, runtime_checkable
from wf_artifacts import ArtifactKind
@dataclass(frozen=True, slots=True)
class TraceRange:
"""Caller-bounded debug trace slice for durable deployment runs."""
start: int = 0
limit: int = 25
@runtime_checkable
class WorkflowApiBackend(Protocol):
"""High-level workflow operation protocol.
Implementations wrap a concrete service (e.g. WorkflowSurfaceHandlers
backed by WfMcpService) so that wf_api never imports wf_mcp.
"""
# -- capabilities --
async def list_capabilities(
self,
*,
query: str | None = None,
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]: ...
async def inspect_capability(
self,
*,
qualified_name: str,
) -> dict[str, Any]: ...
async def call_capability(
self,
*,
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]: ...
# -- artifacts --
async def list_artifacts(
self,
*,
query: str | None = None,
kind: ArtifactKind | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]: ...
async def inspect_artifact(
self,
*,
artifact_id: str,
version: int,
) -> dict[str, Any]: ...
async def save_artifact(
self,
artifact: dict[str, Any],
) -> dict[str, Any]: ...
async def create_artifact_from_plan(
self,
*,
artifact_id: str,
version: int,
title: str,
plan: dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]: ...
async def create_artifact_from_draft(
self,
*,
artifact_id: str,
version: int,
title: str,
draft: dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]: ...
async def create_artifact_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]: ...
async def create_wrapper_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]: ...
# -- drafts --
async def validate_draft(
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]: ...
async def compile_draft(
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]: ...
async def patch_draft(
self,
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]: ...
# -- draft workspaces --
async def list_draft_workspaces(self) -> dict[str, Any]: ...
async def create_draft_workspace(
self,
*,
workspace_id: str,
draft: dict[str, Any],
title: str | None = None,
) -> dict[str, Any]: ...
async def get_draft_workspace(
self,
*,
workspace_id: str,
include_draft: bool = False,
) -> dict[str, Any]: ...
async def delete_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]: ...
async def validate_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]: ...
async def patch_draft_workspace(
self,
*,
workspace_id: str,
revision: int,
patch: list[dict[str, Any]],
) -> dict[str, Any]: ...
async def set_draft_name(
self,
*,
workspace_id: str,
revision: int,
name: str,
) -> dict[str, Any]: ...
async def set_draft_route(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
outcome: str,
target: str,
) -> dict[str, Any]: ...
async def set_step_input_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
input_map: dict[str, str],
) -> dict[str, Any]: ...
async def set_step_output_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
output_map: dict[str, str],
) -> dict[str, Any]: ...
async def create_minimal_draft_workspace(
self,
*,
workspace_id: str,
name: str,
capability_name: str,
input_schema: dict[str, Any],
state_schema: dict[str, Any],
output_schema: dict[str, Any],
input: Sequence[Any] | None = None,
output: Sequence[Any] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
title: str | None = None,
) -> dict[str, Any]: ...
async def create_draft_workspace_from_capability(
self,
*,
workspace_id: str,
capability_name: str,
name: str | None = None,
title: str | None = None,
input_schema: dict[str, Any] | None = None,
state_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
input: Sequence[Any] | None = None,
output: Sequence[Any] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
) -> dict[str, Any]: ...
# -- deployments --
async def list_deployments(self) -> dict[str, Any]: ...
async def inspect_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]: ...
async def save_deployment(
self,
deployment: dict[str, Any],
) -> dict[str, Any]: ...
async def delete_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]: ...
async def validate_deployment(
self,
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]: ...
# -- runs --
async def run_deployment(
self,
*,
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRange | None = None,
) -> dict[str, Any]: ...
async def resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRange | None = None,
) -> dict[str, Any]: ...
async def inspect_run(
self,
*,
run_id: str,
) -> dict[str, Any]: ...
async def read_run_trace(
self,
*,
run_id: str,
trace_range: TraceRange,
) -> dict[str, Any]: ...
src/wf_api/models.py
+9
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from pydantic import BaseModel, Field
@@ -7,6 +8,14 @@ from wf_core import Edge
from wf_core.models.steps import InputBinding, Step
@dataclass(frozen=True, slots=True)
class TraceRange:
"""Caller-bounded debug trace slice for durable deployment runs."""
start: int = 0
limit: int = 25
class RawWorkflowPlan(BaseModel):
"""Raw authoring plan using the same graph step and edge models as core."""
src/wf_api/runs.py
+4 -2
View File
@@ -30,8 +30,10 @@ from .operation_context import WorkflowOperationContext
class TraceRangeLike(Protocol):
"""Small structural trace range accepted from MCP, CLI, or HTTP adapters."""
start: int
limit: int
@property
def start(self) -> int: ...
@property
def limit(self) -> int: ...
class WorkflowRunApi:
src/wf_api/service.py
+55 -45
View File
@@ -5,20 +5,30 @@ from typing import Any
from wf_artifacts import ArtifactKind
from .backend import TraceRange, WorkflowApiBackend
from .artifacts import WorkflowArtifactApi
from .capabilities import WorkflowCapabilityApi
from .deployments import WorkflowDeploymentApi
from .drafts import WorkflowDraftApi
from .models import RawWorkflowPlan
from .operation_context import WorkflowOperationContext
from .runs import TraceRangeLike, WorkflowRunApi
class WorkflowApi:
"""Protocol-neutral workflow application facade.
Delegates every operation to a WorkflowApiBackend implementation.
This class owns no business logic; it exists so that callers
(wf_cli, wf_mcp tools, future HTTP adapters) share one entry point
that does not import wf_mcp.
This facade owns the stable application entry point. It composes the
domain APIs from a WorkflowOperationContext so MCP, CLI, and future HTTP
callers share one operation surface without importing wf_mcp.
"""
def __init__(self, backend: WorkflowApiBackend) -> None:
self.backend: WorkflowApiBackend = backend
def __init__(self, context: WorkflowOperationContext) -> None:
self.context = context
self.capabilities = WorkflowCapabilityApi(context)
self.drafts = WorkflowDraftApi(context)
self.artifacts = WorkflowArtifactApi(context)
self.deployments = WorkflowDeploymentApi(context)
self.runs = WorkflowRunApi(context)
# -- capabilities --
@@ -30,7 +40,7 @@ class WorkflowApi:
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self.backend.list_capabilities(
return await self.capabilities.list_capabilities(
query=query,
source_id=source_id,
cursor=cursor,
@@ -42,7 +52,7 @@ class WorkflowApi:
*,
qualified_name: str,
) -> dict[str, Any]:
return await self.backend.inspect_capability(qualified_name=qualified_name)
return await self.capabilities.inspect_capability(qualified_name=qualified_name)
async def call_capability(
self,
@@ -51,7 +61,7 @@ class WorkflowApi:
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
return await self.backend.call_capability(
return await self.capabilities.call_capability(
qualified_name=qualified_name,
payload=payload,
deployment_id=deployment_id,
@@ -67,7 +77,7 @@ class WorkflowApi:
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self.backend.list_artifacts(
return await self.artifacts.list_artifacts(
query=query,
kind=kind,
cursor=cursor,
@@ -80,7 +90,7 @@ class WorkflowApi:
artifact_id: str,
version: int,
) -> dict[str, Any]:
return await self.backend.inspect_artifact(
return await self.artifacts.inspect_artifact(
artifact_id=artifact_id,
version=version,
)
@@ -89,7 +99,7 @@ class WorkflowApi:
self,
artifact: dict[str, Any],
) -> dict[str, Any]:
return await self.backend.save_artifact(artifact)
return await self.artifacts.save_artifact(artifact)
async def create_artifact_from_plan(
self,
@@ -97,7 +107,7 @@ class WorkflowApi:
artifact_id: str,
version: int,
title: str,
plan: dict[str, Any],
plan: RawWorkflowPlan | dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
@@ -105,7 +115,7 @@ class WorkflowApi:
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_artifact_from_plan(
return await self.artifacts.create_artifact_from_plan(
artifact_id=artifact_id,
version=version,
title=title,
@@ -132,7 +142,7 @@ class WorkflowApi:
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_artifact_from_draft(
return await self.artifacts.create_artifact_from_draft(
artifact_id=artifact_id,
version=version,
title=title,
@@ -159,7 +169,7 @@ class WorkflowApi:
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_artifact_from_workspace(
return await self.artifacts.create_artifact_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
@@ -185,7 +195,7 @@ class WorkflowApi:
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_wrapper_from_workspace(
return await self.artifacts.create_wrapper_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
@@ -204,14 +214,14 @@ class WorkflowApi:
*,
draft: dict[str, Any],
) -> dict[str, Any]:
return await self.backend.validate_draft(draft=draft)
return await self.drafts.validate_draft(draft=draft)
async def compile_draft(
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]:
return await self.backend.compile_draft(draft=draft)
return await self.drafts.compile_draft(draft=draft)
async def patch_draft(
self,
@@ -219,12 +229,12 @@ class WorkflowApi:
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self.backend.patch_draft(draft=draft, patch=patch)
return await self.drafts.patch_draft(draft=draft, patch=patch)
# -- draft workspaces --
async def list_draft_workspaces(self) -> dict[str, Any]:
return await self.backend.list_draft_workspaces()
return await self.drafts.list_draft_workspaces()
async def create_draft_workspace(
self,
@@ -233,7 +243,7 @@ class WorkflowApi:
draft: dict[str, Any],
title: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_draft_workspace(
return await self.drafts.create_draft_workspace(
workspace_id=workspace_id,
draft=draft,
title=title,
@@ -245,7 +255,7 @@ class WorkflowApi:
workspace_id: str,
include_draft: bool = False,
) -> dict[str, Any]:
return await self.backend.get_draft_workspace(
return await self.drafts.get_draft_workspace(
workspace_id=workspace_id,
include_draft=include_draft,
)
@@ -255,14 +265,14 @@ class WorkflowApi:
*,
workspace_id: str,
) -> dict[str, Any]:
return await self.backend.delete_draft_workspace(workspace_id=workspace_id)
return await self.drafts.delete_draft_workspace(workspace_id=workspace_id)
async def validate_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]:
return await self.backend.validate_draft_workspace(workspace_id=workspace_id)
return await self.drafts.validate_draft_workspace(workspace_id=workspace_id)
async def patch_draft_workspace(
self,
@@ -271,7 +281,7 @@ class WorkflowApi:
revision: int,
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self.backend.patch_draft_workspace(
return await self.drafts.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
patch=patch,
@@ -284,7 +294,7 @@ class WorkflowApi:
revision: int,
name: str,
) -> dict[str, Any]:
return await self.backend.set_draft_name(
return await self.drafts.set_draft_name(
workspace_id=workspace_id,
revision=revision,
name=name,
@@ -299,7 +309,7 @@ class WorkflowApi:
outcome: str,
target: str,
) -> dict[str, Any]:
return await self.backend.set_draft_route(
return await self.drafts.set_draft_route(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
@@ -315,7 +325,7 @@ class WorkflowApi:
step_id: str,
input_map: dict[str, str],
) -> dict[str, Any]:
return await self.backend.set_step_input_map(
return await self.drafts.set_step_input_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
@@ -330,7 +340,7 @@ class WorkflowApi:
step_id: str,
output_map: dict[str, str],
) -> dict[str, Any]:
return await self.backend.set_step_output_map(
return await self.drafts.set_step_output_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
@@ -353,7 +363,7 @@ class WorkflowApi:
error_message_source: Any | None = None,
title: str | None = None,
) -> dict[str, Any]:
return await self.backend.create_minimal_draft_workspace(
return await self.drafts.create_minimal_draft_workspace(
workspace_id=workspace_id,
name=name,
capability_name=capability_name,
@@ -384,7 +394,7 @@ class WorkflowApi:
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
) -> dict[str, Any]:
return await self.backend.create_draft_workspace_from_capability(
return await self.capabilities.create_draft_workspace_from_capability(
workspace_id=workspace_id,
capability_name=capability_name,
name=name,
@@ -402,27 +412,27 @@ class WorkflowApi:
# -- deployments --
async def list_deployments(self) -> dict[str, Any]:
return await self.backend.list_deployments()
return await self.deployments.list_deployments()
async def inspect_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
return await self.backend.inspect_deployment(deployment_id=deployment_id)
return await self.deployments.inspect_deployment(deployment_id=deployment_id)
async def save_deployment(
self,
deployment: dict[str, Any],
) -> dict[str, Any]:
return await self.backend.save_deployment(deployment)
return await self.deployments.save_deployment(deployment)
async def delete_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
return await self.backend.delete_deployment(deployment_id=deployment_id)
return await self.deployments.delete_deployment(deployment_id=deployment_id)
async def validate_deployment(
self,
@@ -430,7 +440,7 @@ class WorkflowApi:
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]:
return await self.backend.validate_deployment(
return await self.deployments.validate_deployment(
deployment_id=deployment_id,
live_check=live_check,
)
@@ -442,9 +452,9 @@ class WorkflowApi:
*,
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRange | None = None,
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self.backend.run_deployment(
return await self.runs.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
trace_range=trace_range,
@@ -456,9 +466,9 @@ class WorkflowApi:
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRange | None = None,
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self.backend.resume_run(
return await self.runs.resume_run(
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
@@ -470,15 +480,15 @@ class WorkflowApi:
*,
run_id: str,
) -> dict[str, Any]:
return await self.backend.inspect_run(run_id=run_id)
return await self.runs.inspect_run(run_id=run_id)
async def read_run_trace(
self,
*,
run_id: str,
trace_range: TraceRange,
trace_range: TraceRangeLike,
) -> dict[str, Any]:
return await self.backend.read_run_trace(
return await self.runs.read_run_trace(
run_id=run_id,
trace_range=trace_range,
)
src/wf_cli/commands/runs.py
+1 -1
View File
@@ -8,7 +8,7 @@ import typer
from wf_cli.context import config_path_from_context, load_cli_context
from wf_cli.io import CliInputError, emit_json, parse_json_input
from wf_api.backend import TraceRange
from wf_api import TraceRange
app = typer.Typer(
name="run",
src/wf_cli/context.py
+2 -2
View File
@@ -8,7 +8,7 @@ import typer
from wf_api import WorkflowApi
from wf_mcp.broker import build_service_from_config, load_broker_config
from wf_mcp.broker.service import WfMcpService
from wf_mcp.broker.service.workflow_api_backend import WfMcpWorkflowApiBackend
from wf_mcp.broker.service.workflow_operation_context import context_from_service
@dataclass(frozen=True)
@@ -41,5 +41,5 @@ def load_cli_context(config_path: str | Path) -> CliContext:
return CliContext(
config_path=resolved_config_path,
service=service,
handlers=WorkflowApi(WfMcpWorkflowApiBackend(service)),
handlers=WorkflowApi(context_from_service(service)),
)
src/wf_mcp/broker/service/workflow_api_backend.py
-489
View File
@@ -1,489 +0,0 @@
from __future__ import annotations
from collections.abc import Sequence
from typing import Any
from wf_artifacts import ArtifactKind
from wf_api.backend import TraceRange as ApiTraceRange
from ...workflow_surface.handlers import WorkflowSurfaceHandlers
from ...workflow_surface.models import TraceRange as HandlerTraceRange
from .core import WfMcpService
def _to_handler_trace_range(tr: ApiTraceRange) -> HandlerTraceRange:
return HandlerTraceRange(start=tr.start, limit=tr.limit)
class WfMcpWorkflowApiBackend:
"""Adapt existing WorkflowSurfaceHandlers into WorkflowApiBackend."""
def __init__(self, service: WfMcpService) -> None:
self._handlers = WorkflowSurfaceHandlers(service)
# -- capabilities --
async def list_capabilities(
self,
*,
query: str | None = None,
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._handlers.list_capabilities(
query=query,
source_id=source_id,
cursor=cursor,
limit=limit,
)
async def inspect_capability(
self,
*,
qualified_name: str,
) -> dict[str, Any]:
return await self._handlers.inspect_capability(qualified_name=qualified_name)
async def call_capability(
self,
*,
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
return await self._handlers.call_capability(
qualified_name=qualified_name,
payload=payload,
deployment_id=deployment_id,
)
# -- artifacts --
async def list_artifacts(
self,
*,
query: str | None = None,
kind: ArtifactKind | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._handlers.list_artifacts(
query=query,
kind=kind,
cursor=cursor,
limit=limit,
)
async def inspect_artifact(
self,
*,
artifact_id: str,
version: int,
) -> dict[str, Any]:
return await self._handlers.inspect_artifact(
artifact_id=artifact_id,
version=version,
)
async def save_artifact(
self,
artifact: dict[str, Any],
) -> dict[str, Any]:
return await self._handlers.save_artifact(artifact)
async def create_artifact_from_plan(
self,
*,
artifact_id: str,
version: int,
title: str,
plan: dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_artifact_from_plan(
artifact_id=artifact_id,
version=version,
title=title,
plan=plan,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def create_artifact_from_draft(
self,
*,
artifact_id: str,
version: int,
title: str,
draft: dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_artifact_from_draft(
artifact_id=artifact_id,
version=version,
title=title,
draft=draft,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def create_artifact_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_artifact_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
title=title,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def create_wrapper_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_wrapper_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
title=title,
outcomes=outcomes,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
# -- drafts --
async def validate_draft(
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]:
return await self._handlers.validate_draft(draft=draft)
async def compile_draft(
self,
*,
draft: dict[str, Any],
) -> dict[str, Any]:
return await self._handlers.compile_draft(draft=draft)
async def patch_draft(
self,
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self._handlers.patch_draft(draft=draft, patch=patch)
# -- draft workspaces --
async def list_draft_workspaces(self) -> dict[str, Any]:
return await self._handlers.list_draft_workspaces()
async def create_draft_workspace(
self,
*,
workspace_id: str,
draft: dict[str, Any],
title: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_draft_workspace(
workspace_id=workspace_id,
draft=draft,
title=title,
)
async def get_draft_workspace(
self,
*,
workspace_id: str,
include_draft: bool = False,
) -> dict[str, Any]:
return await self._handlers.get_draft_workspace(
workspace_id=workspace_id,
include_draft=include_draft,
)
async def delete_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]:
return await self._handlers.delete_draft_workspace(workspace_id=workspace_id)
async def validate_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]:
return await self._handlers.validate_draft_workspace(workspace_id=workspace_id)
async def patch_draft_workspace(
self,
*,
workspace_id: str,
revision: int,
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self._handlers.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
patch=patch,
)
async def set_draft_name(
self,
*,
workspace_id: str,
revision: int,
name: str,
) -> dict[str, Any]:
return await self._handlers.set_draft_name(
workspace_id=workspace_id,
revision=revision,
name=name,
)
async def set_draft_route(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
outcome: str,
target: str,
) -> dict[str, Any]:
return await self._handlers.set_draft_route(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
outcome=outcome,
target=target,
)
async def set_step_input_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
input_map: dict[str, str],
) -> dict[str, Any]:
return await self._handlers.set_step_input_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
input_map=input_map,
)
async def set_step_output_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
output_map: dict[str, str],
) -> dict[str, Any]:
return await self._handlers.set_step_output_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
output_map=output_map,
)
async def create_minimal_draft_workspace(
self,
*,
workspace_id: str,
name: str,
capability_name: str,
input_schema: dict[str, Any],
state_schema: dict[str, Any],
output_schema: dict[str, Any],
input: Sequence[Any] | None = None,
output: Sequence[Any] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
title: str | None = None,
) -> dict[str, Any]:
return await self._handlers.create_minimal_draft_workspace(
workspace_id=workspace_id,
name=name,
capability_name=capability_name,
input_schema=input_schema,
state_schema=state_schema,
output_schema=output_schema,
input=input,
output=output,
input_map=input_map,
output_map=output_map,
error_message_source=error_message_source,
title=title,
)
async def create_draft_workspace_from_capability(
self,
*,
workspace_id: str,
capability_name: str,
name: str | None = None,
title: str | None = None,
input_schema: dict[str, Any] | None = None,
state_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
input: Sequence[Any] | None = None,
output: Sequence[Any] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
) -> dict[str, Any]:
return await self._handlers.create_draft_workspace_from_capability(
workspace_id=workspace_id,
capability_name=capability_name,
name=name,
title=title,
input_schema=input_schema,
state_schema=state_schema,
output_schema=output_schema,
input=input,
output=output,
input_map=input_map,
output_map=output_map,
error_message_source=error_message_source,
)
# -- deployments --
async def list_deployments(self) -> dict[str, Any]:
return await self._handlers.list_deployments()
async def inspect_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
return await self._handlers.inspect_deployment(deployment_id=deployment_id)
async def save_deployment(
self,
deployment: dict[str, Any],
) -> dict[str, Any]:
return await self._handlers.save_deployment(deployment)
async def delete_deployment(
self,
*,
deployment_id: str,
) -> dict[str, Any]:
return await self._handlers.delete_deployment(deployment_id=deployment_id)
async def validate_deployment(
self,
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]:
return await self._handlers.validate_deployment(
deployment_id=deployment_id,
live_check=live_check,
)
# -- runs --
async def run_deployment(
self,
*,
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: ApiTraceRange | None = None,
) -> dict[str, Any]:
return await self._handlers.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
trace_range=_to_handler_trace_range(trace_range)
if trace_range is not None
else None,
)
async def resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: ApiTraceRange | None = None,
) -> dict[str, Any]:
return await self._handlers.resume_run(
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
trace_range=_to_handler_trace_range(trace_range)
if trace_range is not None
else None,
)
async def inspect_run(
self,
*,
run_id: str,
) -> dict[str, Any]:
return await self._handlers.inspect_run(run_id=run_id)
async def read_run_trace(
self,
*,
run_id: str,
trace_range: ApiTraceRange,
) -> dict[str, Any]:
return await self._handlers.read_run_trace(
run_id=run_id,
trace_range=_to_handler_trace_range(trace_range),
)
src/wf_mcp/workflow_surface/handlers.py
+11 -482
View File
@@ -1,497 +1,26 @@
from __future__ import annotations
from collections.abc import Sequence
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from wf_artifacts import (
ArtifactKind,
)
from wf_core.models.steps import (
InputBinding,
OutputBinding,
)
from wf_core.paths import GraphSourcePath
from wf_api.artifacts import WorkflowArtifactApi
from wf_api.capabilities import WorkflowCapabilityApi
from wf_api.deployments import WorkflowDeploymentApi
from wf_api.drafts import WorkflowDraftApi
from wf_api.listing import paged_list_payload
from wf_api.models import RawWorkflowPlan
from wf_api.runs import WorkflowRunApi
from wf_api import WorkflowApi
from ..broker.service.workflow_operation_context import context_from_service
from .models import TraceRange
if TYPE_CHECKING:
from ..broker.service import WfMcpService
class WorkflowSurfaceHandlers:
"""Reusable implementation behind MCP workflow artifact tools."""
class WorkflowSurfaceHandlers(WorkflowApi):
"""Compatibility wrapper for old wf_mcp.workflow_surface imports.
New code should construct `WorkflowApi(context_from_service(service))`
directly. This shim keeps tests and legacy broker artifact tools working
while the MCP surface is migrated.
"""
def __init__(self, service: WfMcpService) -> None:
self.service = service
context = context_from_service(service)
self._capabilities = WorkflowCapabilityApi(context)
self._drafts = WorkflowDraftApi(context)
self._artifacts = WorkflowArtifactApi(context)
self._deployments = WorkflowDeploymentApi(context)
self._runs = WorkflowRunApi(context)
super().__init__(context_from_service(service))
async def list_artifacts(
self,
*,
query: str | None = None,
kind: ArtifactKind | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
"""Return compact paged saved artifact summaries.
Saved artifacts can contain full raw workflow plans, so list results
deliberately stay summary-only. Use inspect/run tools for detail.
"""
if self.service.artifact_store is None:
return paged_list_payload("nodes", [], cursor=cursor, limit=limit)
return await self._artifacts.list_artifacts(
query=query,
kind=kind,
cursor=cursor,
limit=limit,
)
async def list_capabilities(
self,
*,
query: str | None = None,
source_id: str | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
"""Return compact paged planner-visible workflow capability summaries."""
return await self._capabilities.list_capabilities(
query=query,
source_id=source_id,
cursor=cursor,
limit=limit,
)
async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
"""Return one planner-visible workflow capability contract."""
return await self._capabilities.inspect_capability(
qualified_name=qualified_name,
)
async def call_capability(
self,
*,
qualified_name: str,
payload: dict[str, Any],
deployment_id: str | None = None,
) -> dict[str, Any]:
"""Execute one planner-visible workflow capability for authoring tests."""
return await self._capabilities.call_capability(
qualified_name=qualified_name,
payload=payload,
deployment_id=deployment_id,
)
async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.save_artifact(artifact)
async def create_artifact_from_plan(
self,
*,
artifact_id: str,
version: int,
title: str,
plan: RawWorkflowPlan | dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.create_artifact_from_plan(
artifact_id=artifact_id,
version=version,
title=title,
plan=plan,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
return await self._drafts.validate_draft(draft=draft)
async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
return await self._drafts.compile_draft(draft=draft)
async def create_artifact_from_draft(
self,
*,
artifact_id: str,
version: int,
title: str,
draft: dict[str, Any],
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.create_artifact_from_draft(
artifact_id=artifact_id,
version=version,
title=title,
draft=draft,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def patch_draft(
self,
*,
draft: dict[str, Any],
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self._drafts.patch_draft(draft=draft, patch=patch)
async def list_draft_workspaces(self) -> dict[str, Any]:
"""Return compact summaries for stored draft workspaces."""
return await self._drafts.list_draft_workspaces()
async def create_draft_workspace(
self,
*,
workspace_id: str,
draft: dict[str, Any],
title: str | None = None,
) -> dict[str, Any]:
return await self._drafts.create_draft_workspace(
workspace_id=workspace_id,
draft=draft,
title=title,
)
async def get_draft_workspace(
self,
*,
workspace_id: str,
include_draft: bool = False,
) -> dict[str, Any]:
return await self._drafts.get_draft_workspace(
workspace_id=workspace_id,
include_draft=include_draft,
)
async def delete_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
return await self._drafts.delete_draft_workspace(workspace_id=workspace_id)
async def validate_draft_workspace(self, *, workspace_id: str) -> dict[str, Any]:
"""Refresh stored validation status without changing draft revision."""
return await self._drafts.validate_draft_workspace(workspace_id=workspace_id)
async def patch_draft_workspace(
self,
*,
workspace_id: str,
revision: int,
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self._drafts.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
patch=patch,
)
async def set_draft_name(
self,
*,
workspace_id: str,
revision: int,
name: str,
) -> dict[str, Any]:
return await self._drafts.set_draft_name(
workspace_id=workspace_id,
revision=revision,
name=name,
)
async def set_draft_route(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
outcome: str,
target: str,
) -> dict[str, Any]:
return await self._drafts.set_draft_route(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
outcome=outcome,
target=target,
)
async def set_step_input_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
input_map: dict[str, str],
) -> dict[str, Any]:
return await self._drafts.set_step_input_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
input_map=input_map,
)
async def set_step_output_map(
self,
*,
workspace_id: str,
revision: int,
step_id: str,
output_map: dict[str, str],
) -> dict[str, Any]:
return await self._drafts.set_step_output_map(
workspace_id=workspace_id,
revision=revision,
step_id=step_id,
output_map=output_map,
)
async def create_minimal_draft_workspace(
self,
*,
workspace_id: str,
name: str,
capability_name: str,
input_schema: dict[str, Any],
state_schema: dict[str, Any],
output_schema: dict[str, Any],
input: Sequence[InputBinding] | None = None,
output: Sequence[OutputBinding] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: str | GraphSourcePath | None = None,
title: str | None = None,
) -> dict[str, Any]:
"""Bootstrap the smallest patchable draft around one workflow capability."""
return await self._drafts.create_minimal_draft_workspace(
workspace_id=workspace_id,
name=name,
capability_name=capability_name,
input_schema=input_schema,
state_schema=state_schema,
output_schema=output_schema,
input=input,
output=output,
input_map=input_map,
output_map=output_map,
error_message_source=error_message_source,
title=title,
)
async def create_draft_workspace_from_capability(
self,
*,
workspace_id: str,
capability_name: str,
name: str | None = None,
title: str | None = None,
input_schema: dict[str, Any] | None = None,
state_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
input: Sequence[InputBinding] | None = None,
output: Sequence[OutputBinding] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: str | GraphSourcePath | None = None,
) -> dict[str, Any]:
"""Create a patchable draft workspace from inspect_capability hints."""
return await self._capabilities.create_draft_workspace_from_capability(
workspace_id=workspace_id,
capability_name=capability_name,
name=name,
title=title,
input_schema=input_schema,
state_schema=state_schema,
output_schema=output_schema,
input=input,
output=output,
input_map=input_map,
output_map=output_map,
error_message_source=error_message_source,
)
async def create_artifact_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.create_artifact_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
title=title,
outcomes=outcomes,
kind=kind,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def create_wrapper_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
description: str | None = None,
required_capabilities: dict[str, dict[str, Any]] | None = None,
source_bindings: dict[str, str] | None = None,
created_from_catalog_version: str | None = None,
) -> dict[str, Any]:
"""Save the current draft workspace as a callable wrapper artifact."""
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.create_wrapper_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
title=title,
outcomes=outcomes,
description=description,
required_capabilities=required_capabilities,
source_bindings=source_bindings,
created_from_catalog_version=created_from_catalog_version,
)
async def inspect_artifact(
self, *, artifact_id: str, version: int
) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._artifacts.inspect_artifact(
artifact_id=artifact_id,
version=version,
)
async def list_deployments(self) -> dict[str, Any]:
if self.service.artifact_store is None:
return {"deployments": []}
return await self._deployments.list_deployments()
async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._deployments.inspect_deployment(
deployment_id=deployment_id,
)
async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._deployments.save_deployment(deployment)
async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
"""Delete one mutable deployment environment binding."""
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._deployments.delete_deployment(
deployment_id=deployment_id,
)
async def validate_deployment(
self,
*,
deployment_id: str,
live_check: bool = False,
) -> dict[str, Any]:
if self.service.artifact_store is None:
raise KeyError("workflow artifact store is not configured")
return await self._deployments.validate_deployment(
deployment_id=deployment_id,
live_check=live_check,
)
async def run_deployment(
self,
*,
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRange | None = None,
) -> dict[str, Any]:
return await self._runs.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
trace_range=trace_range,
)
async def resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRange | None = None,
) -> dict[str, Any]:
"""Resume one durable interrupted deployment run."""
return await self._runs.resume_run(
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
trace_range=trace_range,
)
async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
"""Return one durable stopped-run summary without debug trace entries."""
return await self._runs.inspect_run(run_id=run_id)
async def read_run_trace(
self,
*,
run_id: str,
trace_range: TraceRange,
) -> dict[str, Any]:
"""Return only a caller-bounded debug trace slice from a stopped run."""
return await self._runs.read_run_trace(
run_id=run_id,
trace_range=trace_range,
)
__all__ = ["WorkflowSurfaceHandlers"]
src/wf_mcp/workflow_surface/tools.py
+5 -13
View File
@@ -8,9 +8,8 @@ from pydantic import Field
from wf_artifacts import ArtifactKind
from wf_artifacts.models import RequiredCapability
from wf_api import WorkflowApi
from wf_api.backend import TraceRange as ApiTraceRange
from wf_mcp.broker.service import WfMcpService
from wf_mcp.broker.service.workflow_api_backend import WfMcpWorkflowApiBackend
from wf_mcp.broker.service.workflow_operation_context import context_from_service
from .models import (
CallCapabilityResult,
CreateArtifactFromWorkspaceRequest,
@@ -37,10 +36,7 @@ from .models import (
def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None:
"""Register stable workflow tools on the public MCP server surface."""
handlers = WorkflowApi(WfMcpWorkflowApiBackend(service))
def _to_api_trace_range(tr: TraceRange) -> ApiTraceRange:
return ApiTraceRange(start=tr.start, limit=tr.limit)
handlers = WorkflowApi(context_from_service(service))
@server.tool(
name="wf.workflow.list_artifacts",
@@ -669,9 +665,7 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
await handlers.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
trace_range=_to_api_trace_range(trace_range)
if trace_range is not None
else None,
trace_range=trace_range,
)
)
@@ -703,9 +697,7 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
trace_range=_to_api_trace_range(trace_range)
if trace_range is not None
else None,
trace_range=trace_range,
)
)
@@ -742,6 +734,6 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
return RunDeploymentResult.model_validate(
await handlers.read_run_trace(
run_id=run_id,
trace_range=_to_api_trace_range(trace_range),
trace_range=trace_range,
)
)
tests/wf_api/test_cli_context_uses_api.py
+1 -1
View File
@@ -31,4 +31,4 @@ def test_load_cli_context_returns_workflow_api(tmp_path: Path) -> None:
context = load_cli_context(config_path)
assert isinstance(context.handlers, WorkflowApi)
assert hasattr(context.handlers, "backend")
assert hasattr(context.handlers, "context")
tests/wf_api/test_direct_service.py
+73
View File
@@ -0,0 +1,73 @@
from __future__ import annotations
import asyncio
from wf_artifacts import FileWorkflowArtifactStore
from wf_api import WorkflowApi
from wf_api.artifacts import WorkflowArtifactApi
from wf_api.capabilities import WorkflowCapabilityApi
from wf_api.deployments import WorkflowDeploymentApi
from wf_api.drafts import WorkflowDraftApi
from wf_api.runs import WorkflowRunApi
from wf_mcp.broker import WfMcpService
from wf_mcp.broker.service.workflow_operation_context import context_from_service
from wf_mcp.models import ConnectionConfig
from wf_mcp.storage import FileStore
from tests.wf_mcp.test_support import echo_tool, local_temp_root
def _api() -> WorkflowApi:
root = local_temp_root() / "wf_api_direct_composition"
service = WfMcpService(
store=FileStore(root / "mcp"),
artifact_store=FileWorkflowArtifactStore(root),
)
service.register_connection(
ConnectionConfig(id="demo.personal", server="demo", account="personal")
)
service.register_specs("demo.personal", echo_tool)
return WorkflowApi(context_from_service(service))
def test_workflow_api_composes_domain_services() -> None:
api = _api()
assert isinstance(api.capabilities, WorkflowCapabilityApi)
assert isinstance(api.drafts, WorkflowDraftApi)
assert isinstance(api.artifacts, WorkflowArtifactApi)
assert isinstance(api.deployments, WorkflowDeploymentApi)
assert isinstance(api.runs, WorkflowRunApi)
assert not hasattr(api, "backend")
def test_workflow_api_direct_capability_call() -> None:
api = _api()
result = asyncio.run(
api.call_capability(
qualified_name="demo.personal.echo_tool",
payload={"text": "hello"},
)
)
assert result["kind"] == "node_spec"
assert result["outcome"] == "ok"
assert result["output"] == {"echoed": "hello"}
def test_workflow_surface_handlers_is_compatibility_shim() -> None:
from wf_api import WorkflowApi
from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
root = local_temp_root() / "workflow_surface_handler_shim"
service = WfMcpService(
store=FileStore(root / "mcp"),
artifact_store=FileWorkflowArtifactStore(root),
)
handlers = WorkflowSurfaceHandlers(service)
assert isinstance(handlers, WorkflowApi)
assert handlers.service is service
assert handlers.context.artifact_store is service.artifact_store
tests/wf_api/test_no_double_delegation.py
+28
View File
@@ -0,0 +1,28 @@
from __future__ import annotations
import ast
from pathlib import Path
def _imports_module(path: Path, module_name: str) -> bool:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom) and node.module == module_name:
return True
if isinstance(node, ast.Import):
if any(alias.name == module_name for alias in node.names):
return True
return False
def test_cli_and_mcp_tools_do_not_import_backend_adapter() -> None:
root = Path(__file__).resolve().parents[2]
assert not _imports_module(
root / "src" / "wf_cli" / "context.py",
"wf_mcp.broker.service.workflow_api_backend",
)
assert not _imports_module(
root / "src" / "wf_mcp" / "workflow_surface" / "tools.py",
"wf_mcp.broker.service.workflow_api_backend",
)
tests/wf_api/test_raw_workflow_plan_extraction.py
+9
View File
@@ -18,3 +18,12 @@ def test_canonical_and_compat_are_identical() -> None:
from wf_mcp.models import RawWorkflowPlan as Compat
assert Canonical is Compat
def test_trace_range_exports_from_wf_api_models() -> None:
from wf_api import TraceRange
from wf_api.models import TraceRange as CanonicalTraceRange
assert TraceRange is CanonicalTraceRange
assert TraceRange(start=1, limit=2).start == 1
assert TraceRange(start=1, limit=2).limit == 2
tests/wf_cli/test_context.py
+6 -1
View File
@@ -30,4 +30,9 @@ def test_load_cli_context_builds_service_and_handlers(tmp_path: Path) -> None:
assert context.config_path == config_path
assert context.service.connections.list_all()[0].id == "demo.personal"
assert context.handlers.backend._handlers.service is context.service # type: ignore[attr-defined]
assert context.handlers.context.artifact_store is context.service.artifact_store
assert (
context.handlers.context.draft_workspace_store
is context.service.draft_workspace_store
)
assert context.handlers.context.run_store is context.service.run_store