feat: expose source admin over rpc and cli

This commit is contained in:
lda
2026-06-03 21:49:13 +07:00 Verified
parent 24ca91afb1
commit f0d0cffa17
15 changed files with 246 additions and 22 deletions
+2
View File
@@ -104,6 +104,8 @@ implementation state.
`WorkflowSourceAdminApi` / `WorkflowSourceAdminSurface`; MCP admin source
tools delegate through it while connection/raw MCP operations remain
broker-owned.
- Completed: read-only source inventory is exposed through JSON-RPC HTTP and
`wf source list` / `wf source inspect`.
5. **CLI/API alignment**
- Completed for the basic lifecycle: selected `wf` commands can target local
@@ -80,12 +80,13 @@ surface, or plain local CLI utilities.
## Next Slices
1. **Source/admin transport and CLI commands**
- Build JSON-RPC methods and `wf source ...` commands over
`WorkflowSourceAdminSurface`.
1. **Store-backed source registry**
- Read-only source/admin operations are now available through JSON-RPC HTTP
and `wf source list` / `wf source inspect`.
- Next source work is persistence for server-owned dynamic source changes.
- Keep mutation out until the store-backed source registry is designed.
2. **Store-backed source registry**
2. **Mutable source/admin commands**
- Config can bootstrap sources, but server-owned dynamic source changes
should persist through the store.
- Keep source identity structural: source id, provider/account/profile, and
+12 -1
View File
@@ -4,7 +4,17 @@ from typing import Annotated
import typer
from .commands import artifacts, caps, deployments, docs, drafts, explain, runs, schema
from .commands import (
artifacts,
caps,
deployments,
docs,
drafts,
explain,
runs,
schema,
sources,
)
from .context import CliTyperState
app = typer.Typer(
@@ -51,6 +61,7 @@ app.add_typer(drafts.app, name="draft")
app.add_typer(artifacts.app, name="artifact")
app.add_typer(deployments.app, name="deploy")
app.add_typer(runs.app, name="run")
app.add_typer(sources.app, name="source")
app.add_typer(docs.app, name="docs")
app.add_typer(schema.app, name="schema")
app.command("explain")(explain.explain_command)
+52
View File
@@ -0,0 +1,52 @@
from __future__ import annotations
import asyncio
from typing import Annotated
import typer
from wf_cli.context import load_cli_context_from_typer
from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import emit_json
app = typer.Typer(
name="source",
help="List and inspect workflow capability sources.",
no_args_is_help=True,
)
@app.command("list")
def list_sources(
ctx: typer.Context,
cursor: Annotated[
str | None, typer.Option("--cursor", help="Pagination cursor.")
] = None,
limit: Annotated[
int, typer.Option("--limit", min=1, max=100, help="Maximum rows.")
] = 50,
output_format: Annotated[
ListOutputFormat, typer.Option("--format", help="Output format.")
] = ListOutputFormat.JSON,
) -> None:
"""List compact workflow source summaries."""
context = load_cli_context_from_typer(ctx)
payload = asyncio.run(context.source_admin.list_sources(cursor=cursor, limit=limit))
emit_list_payload(
payload,
collection_key="sources",
output_format=output_format,
id_field="id",
summary_fields=("kind", "enabled", "description"),
)
@app.command("inspect")
def inspect_source(
ctx: typer.Context,
source_id: Annotated[str, typer.Argument(help="Workflow source id.")],
) -> None:
"""Inspect one workflow source inventory."""
context = load_cli_context_from_typer(ctx)
payload = asyncio.run(context.source_admin.inspect_source(source_id=source_id))
emit_json(payload)
+23 -11
View File
@@ -8,7 +8,12 @@ from collections.abc import Mapping
import typer
from pydantic import ValidationError
from wf_api import WorkflowApi, WorkflowApiSurface
from wf_api import (
WorkflowApi,
WorkflowApiSurface,
WorkflowSourceAdminApi,
WorkflowSourceAdminSurface,
)
from wf_config import (
FilesystemStoreConfig,
LocalTargetConfig,
@@ -29,6 +34,7 @@ class CliContext:
config_path: Path
service: WfMcpService | None
handlers: WorkflowApiSurface
source_admin: WorkflowSourceAdminSurface
@dataclass(frozen=True)
@@ -98,16 +104,18 @@ def load_cli_context(
if rpc_url is not None:
_validate_rpc_url(rpc_url)
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=RpcWorkflowApiClient(
client = RpcWorkflowApiClient(
url=rpc_url,
timeout_seconds=_rpc_timeout_from_optional_config(
resolved_config_path,
override=rpc_timeout_seconds,
),
),
)
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=client,
source_admin=client,
)
if _is_legacy_mcp_config(resolved_config_path):
@@ -117,6 +125,7 @@ def load_cli_context(
config_path=resolved_config_path,
service=service,
handlers=WorkflowApi(context_from_service(service)),
source_admin=WorkflowSourceAdminApi(context_from_service(service)),
)
config = load_workflow_config(resolved_config_path)
@@ -130,19 +139,22 @@ def load_cli_context(
config_path=resolved_config_path,
service=None,
handlers=server.api,
source_admin=server.source_admin,
)
if isinstance(target, RpcHttpTargetConfig):
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=RpcWorkflowApiClient(
client = RpcWorkflowApiClient(
url=str(target.url),
timeout_seconds=(
rpc_timeout_seconds
if rpc_timeout_seconds is not None
else target.timeout_seconds
),
),
)
return CliContext(
config_path=resolved_config_path,
service=None,
handlers=client,
source_admin=client,
)
raise ValueError(f"unsupported workflow target {target!r}")
+4 -1
View File
@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from wf_api import WorkflowApi, durable_workflow_api
from wf_api import WorkflowApi, WorkflowSourceAdminApi, durable_workflow_api
from wf_api.local_sources import builtin_sources, get_qualified_spec
from wf_api.models import RawWorkflowPlan, TraceRange
from wf_api.operation_context import (
@@ -237,6 +237,7 @@ class WorkflowServer:
stores: WorkflowStores
context: WorkflowOperationContext
api: WorkflowApi
source_admin: WorkflowSourceAdminApi
events: InMemoryWorkflowEventRecorder
@staticmethod
@@ -264,10 +265,12 @@ def build_local_static_workflow_server(root: str | Path) -> WorkflowServer:
live_sources=None,
)
api = durable_workflow_api(context)
source_admin = WorkflowSourceAdminApi(context)
return WorkflowServer(
config=config,
stores=stores,
context=context,
api=api,
source_admin=source_admin,
events=events,
)
+4
View File
@@ -14,10 +14,12 @@ from .models import (
InspectCapabilityParams,
InspectDeploymentParams,
InspectRunParams,
InspectSourceParams,
ListArtifactsParams,
ListCapabilitiesParams,
ListDeploymentsParams,
ListDraftWorkspacesParams,
ListSourcesParams,
PatchDraftParams,
PatchDraftWorkspaceParams,
ReadRunTraceParams,
@@ -42,10 +44,12 @@ __all__ = [
"InspectCapabilityParams",
"InspectDeploymentParams",
"InspectRunParams",
"InspectSourceParams",
"ListArtifactsParams",
"ListCapabilitiesParams",
"ListDeploymentsParams",
"ListDraftWorkspacesParams",
"ListSourcesParams",
"PatchDraftParams",
"PatchDraftWorkspaceParams",
"ReadRunTraceParams",
+2
View File
@@ -12,6 +12,7 @@ from .methods_capabilities import register_methods as register_capability_method
from .methods_deployments import register_methods as register_deployment_methods
from .methods_drafts import register_methods as register_draft_methods
from .methods_runs import register_methods as register_run_methods
from .methods_sources import register_methods as register_source_methods
def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc.API:
@@ -43,6 +44,7 @@ def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc
register_artifact_methods(entrypoint, server)
register_deployment_methods(entrypoint, server)
register_run_methods(entrypoint, server)
register_source_methods(entrypoint, server)
app.bind_entrypoint(entrypoint)
return app
+2
View File
@@ -10,6 +10,7 @@ from .client_capabilities import RpcCapabilityClientMixin
from .client_deployments import RpcDeploymentClientMixin
from .client_drafts import RpcDraftClientMixin
from .client_runs import RpcRunClientMixin
from .client_sources import RpcSourceAdminClientMixin
@dataclass(slots=True)
@@ -20,6 +21,7 @@ class RpcWorkflowApiClient(
RpcArtifactClientMixin,
RpcDeploymentClientMixin,
RpcRunClientMixin,
RpcSourceAdminClientMixin,
):
"""WorkflowApiSurface implementation backed by JSON-RPC HTTP calls.
@@ -0,0 +1,29 @@
from __future__ import annotations
from typing import Any
class RpcSourceAdminClientMixin:
"""JSON-RPC implementation of read-only source admin surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
async def list_sources(
self,
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.sources.list",
{
"cursor": cursor,
"limit": limit,
},
)
async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
return await self._call(
"workflow.sources.inspect",
{"source_id": source_id},
)
@@ -0,0 +1,40 @@
from __future__ import annotations
from typing import Any
from fastapi import Body
import fastapi_jsonrpc as jsonrpc
from fastapi_jsonrpc import Params
from wf_server import WorkflowServer
from .errors import WorkflowRpcError, raise_workflow_rpc_error
from .models import InspectSourceParams, ListSourcesParams
def register_methods(
entrypoint: jsonrpc.Entrypoint,
server: WorkflowServer,
) -> None:
"""Register read-only source/admin JSON-RPC methods."""
@entrypoint.method(name="workflow.sources.list", errors=[WorkflowRpcError])
async def workflow_sources_list(
params: ListSourcesParams = Body(default_factory=ListSourcesParams),
) -> dict[str, Any]:
try:
return await server.source_admin.list_sources(
cursor=params.cursor,
limit=params.limit,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.sources.inspect", errors=[WorkflowRpcError])
async def workflow_sources_inspect(
params: InspectSourceParams = Params(...), # type: ignore[reportArgumentType]
) -> dict[str, Any]:
try:
return await server.source_admin.inspect_source(source_id=params.source_id)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
+9
View File
@@ -37,6 +37,15 @@ class ListCapabilitiesParams(RpcParamsModel):
limit: int = Field(default=50, ge=1, le=200)
class ListSourcesParams(RpcParamsModel):
cursor: str | None = Field(default=None)
limit: int = Field(default=50, ge=1, le=100)
class InspectSourceParams(RpcParamsModel):
source_id: str = Field(min_length=1)
class InspectCapabilityParams(RpcParamsModel):
qualified_name: str = Field(min_length=1)
+9
View File
@@ -17,6 +17,7 @@ def test_wf_help_lists_lifecycle_groups() -> None:
assert "artifact" in result.output
assert "deploy" in result.output
assert "run" in result.output
assert "source" in result.output
assert "docs" in result.output
assert "schema" in result.output
assert "explain" in result.output
@@ -67,6 +68,14 @@ def test_wf_cap_list_help_exists() -> None:
assert "--source" in result.output
def test_wf_source_list_help_exists() -> None:
result = runner.invoke(app, ["source", "list", "--help"])
assert result.exit_code == 0
assert "--format" in result.output
assert "--limit" in result.output
def test_wf_artifact_list_help_exists() -> None:
result = runner.invoke(app, ["artifact", "list", "--help"])
+24
View File
@@ -286,6 +286,30 @@ def test_wf_cap_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
assert '"name": "wf.std.constant"' in result.output
def test_wf_source_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
original_client = httpx.AsyncClient
monkeypatch.setattr(
"wf_transport_rpc_http.client.httpx.AsyncClient",
lambda *args, **kwargs: original_client(
transport=httpx.ASGITransport(app=create_rpc_app(server)),
base_url="http://test",
),
)
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
runner = CliRunner()
base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
listed = runner.invoke(app, [*base_args, "source", "list", "--limit", "10"])
inspected = runner.invoke(app, [*base_args, "source", "inspect", "wf.std"])
assert listed.exit_code == 0, listed.output
assert '"id": "wf.std"' in listed.output
assert inspected.exit_code == 0, inspected.output
assert '"id": "wf.std"' in inspected.output
def test_wf_remote_draft_artifact_deploy_lifecycle(monkeypatch, tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
original_client = httpx.AsyncClient
@@ -83,6 +83,30 @@ def test_rpc_workflow_client_lists_and_inspects_capabilities(tmp_path) -> None:
asyncio.run(scenario())
def test_rpc_workflow_client_lists_and_inspects_sources(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(
transport=transport,
base_url="http://test",
) as http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc",
timeout_seconds=5,
http_client=http_client,
)
listed = await client.list_sources(limit=10)
inspected = await client.inspect_source(source_id="wf.std")
source_ids = {source["id"] for source in listed["sources"]}
assert "wf.std" in source_ids
assert inspected["id"] == "wf.std"
asyncio.run(scenario())
def test_rpc_workflow_client_runs_and_reads_trace(tmp_path) -> None:
async def scenario() -> None:
server = build_local_static_workflow_server(tmp_path / "store")