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

remove just about everything bad coming from broker

Browse Source
This commit is contained in:
lda
2026-05-19 23:35:08 +07:00 Verified
parent 53a9e7f85d
commit 5bda3e3770
19 changed files with 99 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_mcp/broker/prompts.py
+2 -2
View File
@@ -10,8 +10,8 @@ Build workflows from current capabilities instead of assuming a stale catalog.
Use `get_planner_catalog` when you need the current workflow-capability view.
Use `list_sources` when you need to understand source ownership, visibility,
and capability kinds.
Use `call_broker_tool` to test an upstream MCP tool manually before wrapping it
into a workflow.
Use source-projected workflow capabilities or directly exposed proxy tools to
test the smallest reusable piece before wrapping it into a workflow.
Prefer namespaced capabilities, inspect before you rely on them, and test the
smallest reusable piece before saving a larger workflow artifact.
src/wf_mcp/broker/service/builtins.py
+7 -94
View File
@@ -1,30 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Protocol
from typing import TYPE_CHECKING, Any
from pydantic import BaseModel, Field
from wf_authoring import (
NodeReturn,
NodeSpec,
coalesce,
constant,
default_if_none,
first_item,
first_item_maybe,
first_item_or_none,
is_empty,
last_item,
last_item_or_none,
length,
node,
pick_key,
pick_path,
project_fields,
rename_fields,
runtime_error,
truthy,
)
from wf_authoring import NodeSpec, coalesce, constant, default_if_none, first_item
from wf_authoring import first_item_maybe, first_item_or_none, is_empty, last_item
from wf_authoring import last_item_or_none, length, node, pick_key, pick_path
from wf_authoring import project_fields, rename_fields, runtime_error, truthy
from wf_core.runtime.ops.merges import DEFAULT_REDUCER_DEFINITIONS
from wf_platform import (
@@ -42,7 +23,7 @@ BUILTIN_CONNECTION_ID = "wf.std"
"""Internal source id for workflow standard-library node specs."""
MCP_SOURCE_ID = "wf.mcp"
"""Internal source id for broker MCP utility node specs."""
"""Reserved source id for future workflow-safe MCP utility node specs."""
AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
@@ -66,43 +47,6 @@ AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
"""Existing authoring ops that are also exposed through the workflow stdlib."""
class ToolCaller(Protocol):
"""Small service boundary needed by the broker-local MCP utility nodes."""
async def call_tool(
self,
connection_id: str,
tool_name: str,
*,
arguments: dict[str, Any] | None = None,
) -> dict[str, Any]: ...
class McpCallToolInput(BaseModel):
"""Input for calling a proxied MCP tool from inside a workflow."""
connection_id: str = Field(description="Connection id that owns the MCP tool.")
tool_name: str = Field(description="Local tool name on the upstream MCP server.")
arguments: dict[str, Any] = Field(
default_factory=dict,
description="JSON-compatible arguments passed to the upstream tool.",
)
class McpCallToolOutput(BaseModel):
"""Normalized output returned by a proxied MCP tool call."""
outcome: str = Field(description="Workflow outcome reported by the upstream tool.")
output: dict[str, Any] = Field(
default_factory=dict,
description="JSON-compatible tool result payload.",
)
meta: dict[str, Any] = Field(
default_factory=dict,
description="Adapter metadata returned with the tool result.",
)
def builtin_specs() -> dict[str, NodeSpec[Any, Any]]:
"""Return built-in NodeSpecs available to raw broker workflow plans."""
specs = [
@@ -126,30 +70,7 @@ def builtin_reducer_definitions():
return dict(DEFAULT_REDUCER_DEFINITIONS)
def mcp_specs(service: ToolCaller) -> dict[str, NodeSpec[Any, Any]]:
"""Return service-bound MCP utility specs available to raw plans."""
@node(
name="call_tool",
outcomes=("ok", "error"),
input_model=McpCallToolInput,
output_model=McpCallToolOutput,
description="Call a tool on a registered MCP connection.",
)
async def call_tool(payload: McpCallToolInput) -> NodeReturn[McpCallToolOutput]:
result = await service.call_tool(
payload.connection_id,
payload.tool_name,
arguments=payload.arguments,
)
output = McpCallToolOutput.model_validate(result)
return NodeReturn(outcome=output.outcome, output=output)
qualified_specs = [qualify_spec(MCP_SOURCE_ID, call_tool)]
return {spec.name: spec for spec in qualified_specs}
def builtin_sources(service: ToolCaller) -> dict[str, CapabilitySource]:
def builtin_sources() -> dict[str, CapabilitySource]:
"""Return all broker-local capability sources."""
return {
BUILTIN_CONNECTION_ID: CapabilitySource(
@@ -168,12 +89,4 @@ def builtin_sources(service: ToolCaller) -> dict[str, CapabilitySource]:
permissions=SourcePermissions(safe_for_workflow=True),
description="Workflow standard-library nodes.",
),
MCP_SOURCE_ID: CapabilitySource(
id=MCP_SOURCE_ID,
kind="system",
capabilities=CapabilityBuckets(node_specs=mcp_specs(service)),
visibility=SourceVisibility(planner=True, admin_dashboard=True),
permissions=SourcePermissions(calls_upstream=True),
description="Broker MCP utility nodes.",
),
}
src/wf_mcp/broker/service/core.py
+16 -42
View File
@@ -84,7 +84,7 @@ class WfMcpService:
_store_root(self.store)
)
if self.include_builtin_specs:
for source in builtin_sources(self).values():
for source in builtin_sources().values():
self.register_capability_source(source)
self.register_capability_source(admin_source())
@@ -449,41 +449,6 @@ class WfMcpService:
)
return result
async def call_tool(
self,
connection_id: str,
tool_name: str,
*,
arguments: dict[str, Any] | None = None,
) -> dict[str, Any]:
connection = self.connections.get(connection_id)
adapter = require_adapter(connection, self.adapters)
auth = self.load_auth(connection_id)
capability_id = qualify_node_name(connection_id, tool_name)
payload = arguments or {}
self._record_event(
make_event(
"tool_call_started",
connection_id=connection_id,
capability_id=capability_id,
payload={"argument_keys": sorted(payload.keys())},
)
)
result = await adapter.call_tool(connection, auth, tool_name, payload)
self._record_event(
make_event(
"tool_call_completed",
connection_id=connection_id,
capability_id=capability_id,
payload={"outcome": result.outcome},
)
)
return {
"outcome": result.outcome,
"output": result.output,
"meta": result.meta,
}
async def send_notification(
self,
connection_id: str,
@@ -794,20 +759,29 @@ class WfMcpService:
self,
entry: CatalogNodeEntry,
) -> NodeSpec[Any, Any]:
"""Rebuild an executable tool wrapper from a stored catalog node entry."""
"""Rebuild an executable tool wrapper from a stored catalog node entry.
Snapshot entries store schema/name metadata, not Python functions. This
helper reconstructs the same generated NodeSpec shape and routes calls
through `_tool_executor_for()`, so hydrated specs use the persistent MCP
runtime when the service has one configured.
"""
model_prefix = entry.qualified_name.replace(".", "_").replace("-", "_")
input_model = _model_from_schema(f"{model_prefix}_Input", entry.input_schema)
output_model = _model_from_schema(f"{model_prefix}_Output", entry.output_schema)
async def invoke_tool(payload: BaseModel) -> NodeReturn[BaseModel]:
result = await self.call_tool(
entry.connection_id,
connection = self.connections.get(entry.connection_id)
auth = self.load_auth(entry.connection_id)
result = await self._tool_executor_for(connection).call_tool(
connection,
auth,
entry.local_name,
arguments=payload.model_dump(),
payload.model_dump(exclude_unset=True),
)
return NodeReturn(
outcome=result["outcome"],
output=output_model.model_validate(result["output"]),
outcome=result.outcome,
output=output_model.model_validate(result.output),
)
return NodeSpec(