Csf Strategy¶

csf_strategy — mvp.csf_strategy

Cluster: Safety & Alignment | Type: component | MCP Tools: 26

Overview¶

Strategy registry and execution block that manages named safety strategies: composable plans for how an agent should behave under specific conditions. Strategies are registered with strategy_register, retrieved with strategy_get, approved with strategy_approve in the MCP surface, and executed via strategy_run, which delegates to a StrategyRunner that optionally calls an LLM for adaptive plan generation. The returns library is used for railway-oriented error propagation through the dispatch chain.

When to use:

• Registering and running safety-aware behavioural strategies for an agent at runtime
• Composing multi-step safety plans from reusable strategy primitives
• Switching between pre-registered strategies based on current context or risk level
• Capturing strategy execution traces, audit history, and simple before/after comparisons

Example:

from mvp.csf_strategy.strategy_mcp import CSFStrategyMCPBlock, MCPStrategyInput

block = CSFStrategyMCPBlock(db_path=":memory:")
block.infer(MCPStrategyInput(
    op="strategy_register",
    strategy_id="conservative",
    description="Check, execute, verify, and critique a task.",
    capabilities=["echo"],
))
block.infer(MCPStrategyInput(op="strategy_approve", strategy_id="conservative"))
result = block.infer(MCPStrategyInput(
    op="strategy_run",
    strategy_id="conservative",
    task="Check the generated summary for obvious omissions.",
))
# result.value.value -> {"run_id": ..., "success": ..., "nodes_executed": ...}

Works well with: csf, csf_cognitive, goal_engine

Public API¶

Node(ABC)¶

Base execution node in a strategy graph.

Constructor:

Methods:

execute(context: dict[str, Any], tools: dict[str, Any]) -> dict[str, Any]¶

Execute this node, updating and returning the context.

PlannerNode(Node)¶

Decomposes a task into a plan (list of subtask strings).

Constructor:

Methods:

execute(context: dict[str, Any], tools: dict[str, Any]) -> dict[str, Any]¶

ExecutorNode(Node)¶

Executes each step in the plan using available tools.

Methods:

execute(context: dict[str, Any], tools: dict[str, Any]) -> dict[str, Any]¶

VerifierNode(Node)¶

Verifies execution results for errors and completeness.

Methods:

execute(context: dict[str, Any], tools: dict[str, Any]) -> dict[str, Any]¶

CriticNode(Node)¶

Evaluates quality and suggests improvements.

Constructor:

Methods:

execute(context: dict[str, Any], tools: dict[str, Any]) -> dict[str, Any]¶

ExecutionGraph¶

Ordered sequence of nodes executed in pipeline fashion.

Constructor:

Methods:

add_node(node: Node) -> None¶

run(task: str, tools: dict[str, Any]) -> dict[str, Any]¶

Execute all nodes sequentially, building up the context.

build_default(llm: Any = None) -> ExecutionGraph¶

Build the standard Planner->Executor->Verifier->Critic graph.

build_custom(node_names: list[str], llm: Any = None) -> ExecutionGraph¶

Build a graph from a list of node type names.

StrategyStats¶

Aggregate statistics for a strategy.

TraceStore¶

In-memory trace storage.

Methods:

record(trace: dict[str, Any]) -> str¶

Store a trace and return its ID.

list_for_strategy(strategy_id: str, limit: int = 50) -> list[dict[str, Any]]¶

List traces for a given strategy.

get(trace_id: str) -> dict[str, Any] | None¶

Get a single trace by ID.

all_traces() -> list[dict[str, Any]]¶

Return all traces.

TraceLearner¶

Computes statistics and proposes recipe adjustments from traces.

Constructor:

Methods:

compute_stats(strategy_id: str) -> StrategyStats¶

Compute aggregate statistics for a strategy from traces.

propose_adjustments(strategy_id: str) -> list[dict[str, Any]]¶

Propose recipe adjustments based on trace analysis.

compare_strategies(strategy_a: str, strategy_b: str) -> dict[str, Any]¶

Compare two strategies' performance statistics.

ApprovalStatus(str, Enum)¶

RetryPolicy¶

FailureMode¶

EvaluationMetric¶

StrategyModel¶

Methods:

to_dict() -> dict[str, Any]¶

from_dict(d: dict[str, Any]) -> StrategyModel¶

StrategyRegistry¶

Thread-safe registry with versioning, audit trail, and optional SQLite persistence.

Constructor:

Methods:

register(model: StrategyModel) -> int¶

Register a strategy. Returns the assigned version number.

load(model: StrategyModel) -> None¶

Load a persisted strategy into memory without changing its metadata.

get(sid: str, version: int | None = None) -> StrategyModel | None¶

Get latest version or a specific version of a strategy.

approve(sid: str, reviewer: str | None = None) -> bool¶

Approve the latest version of a strategy with reviewer binding.

deprecate(sid: str) -> bool¶

Deprecate the latest version of a strategy.

list_all() -> list[dict[str, Any]]¶

List all registered strategies with summary metadata.

get_history(sid: str) -> list[dict[str, Any]]¶

Return all versions of a strategy.

get_audit_log() -> list[dict[str, Any]]¶

Return the full audit trail.

StrategyRunner¶

Executes strategies against tasks using the execution graph.

Constructor:

Methods:

check_capabilities(strategy_id: str) -> dict[str, Any]¶

Pre-flight check: verify all required capabilities are available.

run(strategy_id: str, task: str) -> dict[str, Any]¶

Execute a strategy: build graph, run all nodes, return context.

dry_run(strategy_id: str, task: str) -> dict[str, Any]¶

Dry-run with EchoTools substituted for all tools.

run_node(strategy_id: str, node_name: str, task: str, context: dict[str, Any] | None = None) -> dict[str, Any]¶

Run a single named node.

StrategyInput(BaseModel)¶

StrategyOutput(BaseModel)¶

CSFStrategyBlock(AIBlock[StrategyInput, StrategyOutput, dict])¶

Base strategy block with 5 core operations.

Methods:

infer(data: StrategyInput) -> Result[StrategyOutput]¶

SourceOperation¶

Typed descriptor for a backend-native source operation a tool can expose.

Methods:

compact() -> dict[str, Any]¶

SourceOperationResult¶

Structured result of a source-operation invocation.

Methods:

compact() -> dict[str, Any]¶

Tool(ABC)¶

Abstract base class for strategy tools.

Constructor:

Methods:

execute(input_text: str) -> str¶

Execute the tool on the given input and return a string result.

source_operations() -> list[SourceOperation]¶

Describe the backend-native operations this tool exposes.

execute_operation(op_name: str, payload: dict[str, Any] | None = None, context: dict[str, Any] | None = None) -> SourceOperationResult¶

Execute a named source operation.

EchoTool(Tool)¶

Passes input through unchanged. Used for dry-run testing.

Constructor:

Methods:

execute(input_text: str) -> str¶

source_operations() -> list[SourceOperation]¶

LLMTool(Tool)¶

Calls the claude CLI subprocess for LLM-powered operations.

Constructor:

Methods:

execute(input_text: str) -> str¶

ToolRegistry¶

Registry of named tools with capability lookup.

Methods:

register(tool: Tool) -> None¶

Register a tool by name.

get(name: str) -> Tool | None¶

Get a tool by name.

list_all() -> list[dict[str, Any]]¶

List all registered tools with metadata.

get_capabilities() -> list[str]¶

Return the union of all tool capabilities.

as_dict() -> dict[str, Tool]¶

Return tools as a name->Tool dict (for graph execution).

source_operations() -> list[dict[str, Any]]¶

Discover all backend-native source operations across registered tools.

find_by_source_operation(op_name: str) -> Tool | None¶

Return the first tool that declares a source operation named op_name.

find_by_capability(capability: str) -> list[Tool]¶

Return all tools whose capabilities include capability.

MCP Tools¶