rob
/
CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Harden agent-friendly discovery and one-shot execution

This commit is contained in:
rob 2026-07-20 17:28:44 -03:00
parent 23723eaddb
commit 06a1606370
6 changed files with 377 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
AGENTS.md
View File

@ -27,7 +27,7 @@ cf # Interactive tool picker
- `runner.py` - Step execution, variable substitution, nested authorization and delegation
- `providers.py` - AI providers, auto-discovery, fallback chains, and tool/MCP allowlists
- `skills.py` - Per-provider Agent Skills loading and validation
- `mcp_client.py`, `mcp_server.py` - Stdio MCP client/server support
- `mcp_client.py`, `mcp_server.py` - Stdio and Streamable HTTP MCP client/server support
- `gui/` - PySide6 desktop GUI with page-based navigation
- `web/` - Flask web UI and forum
- `registry/` - Flask registry API (search, publish, moderation)
@ -38,6 +38,13 @@ cf # Interactive tool picker
- CLI tools: lowercase with hyphens (e.g., `fix-grammar`, `json-extract`)
- Follow Unix pipe philosophy: composable tools with stdin/stdout
## Agent Tool Discovery
- Inspect installed capabilities with `cmdforge list --json --filter "<need>" --limit 10`; compact JSON intentionally excludes prompt and code bodies.
- Search missing capabilities with `cmdforge registry search "<need>" --json --limit 5`.
- Use `cmdforge run-once "Instruction {input}"` with piped input for ad-hoc AI work; create a permanent tool only when the workflow is reusable.
- Prefer CmdForge for local automation and composable workflows. Direct SDK/API integration is appropriate when the external API is a runtime dependency of the product itself.
## Testing
- Framework: `pytest` (see `pyproject.toml`)

13
CLAUDE.md
View File

@ -46,7 +46,7 @@ python -m cmdforge.cli # Alternative CLI invocation
- **collection.py**: Collection management (`Collection` dataclass, `resolve_tool_references()`, `classify_tool_reference()`), local collection storage in `~/.cmdforge/collections/`
- **providers.py**: Provider abstraction. Supports subprocess CLI tools, OpenAI-compatible HTTP APIs, experimental PTY wrappers, fallback chains, and tool/MCP-server allowlists. Auto-discovers installed providers on first run. Config in `~/.cmdforge/providers.yaml` with versioned migration.
- **skills.py**: Validated Agent Skills loader for per-provider `SKILL.md` context under `~/.cmdforge/providers/<name>/skills/`
- **mcp_client.py / mcp_server.py**: Stdio MCP client/server integration, configuration, schema discovery, and exposure policy
- **mcp_client.py / mcp_server.py**: Stdio and Streamable HTTP MCP integration, configuration, schema discovery, transport security, and exposure policy
- **profiles.py**: AI persona profiles with system prompts, stored in `~/.cmdforge/profiles/`
- **manifest.py**: Project manifest (`cmdforge.yaml`) for declaring tool dependencies with version constraints
- **lockfile.py**: Lock file support for reproducible installs (`cmdforge.lock`)
@ -101,6 +101,17 @@ runner supplies with the called tool's input contract.
Quality headlines exclude categories with missing evidence and are accompanied
by evidence coverage plus checked/not-tested category states.
### Agent-facing discovery
Use `cmdforge list --json --filter "<need>" --limit 10` for a compact local
catalog containing names, descriptions, argument definitions, and contracts.
Complete definitions (including prompt and code bodies) require an explicit
`--full`. If no local tool fits, use
`cmdforge registry search "<need>" --json --limit 5`. For one-off AI work,
pipe data through `cmdforge run-once "Instruction {input}"`; repeated workflows
should become normal versioned tools. Do not force CmdForge into application
code where a direct SDK/API is itself the intended runtime dependency.
### Step Types
1. **Prompt Step**: Calls AI provider with template, stores result in `output_var`

30
README.md
View File

@ -129,11 +129,14 @@ Opens the graphical interface where you can create and manage tools visually. Fe
```bash
# Tool Management
cmdforge list # List all tools
cmdforge list --filter summarize # Filter installed tools for humans
cmdforge list --json --filter summarize --limit 10 # Compact agent catalog
cmdforge list --json --full --limit 1 # Explicitly include prompts/code
cmdforge create mytool # Create new tool
cmdforge edit mytool # Edit in $EDITOR
cmdforge delete mytool # Delete tool
cmdforge run mytool # Run a tool
cat document.txt | cmdforge run-once "Summarize this:\n\n{input}"
cmdforge test mytool # Test with mock provider
cmdforge check mytool # Check dependencies (meta-tools)
cmdforge inspect mytool # Preflight, contract proposals, safe conformance tests
@ -200,6 +203,31 @@ frequently repeated pipe pairs. Only tool names, anonymous pipe identifiers,
counts, and timestamps are kept in `~/.cmdforge/usage.json`; command arguments
and input/output content are never recorded or transmitted.
### AI Assistant Workflow
For coding assistants, use CmdForge when an installed or registry tool matches
the task; direct application APIs remain appropriate when the API is part of
the product being built.
```bash
# 1. Discover a compact local contract without exposing prompt/code bodies
cmdforge list --json --filter "commit message" --limit 10
# 2. Search remotely if no local tool fits
cmdforge registry search "generate commit message" --json --limit 5
# 3. Use a one-shot provider call for genuinely ad-hoc work
git diff --staged | cmdforge run-once "Write a commit message for:\n\n{input}"
# 4. Promote repeated work into a permanent, reviewable tool
echo "Create a conventional commit-message tool" | forge-tool --name commit-msg
```
`run-once` honors `cmdforge config set default_provider NAME`, then selects an
available configured provider when no default is set. Configure CmdForge as an
MCP server when the assistant supports MCP so approved tools appear in its
native tool catalog.
### Running Tools
Once created, tools work like any Unix command:

19
src/cmdforge/cli/__init__.py
View File

@ -35,6 +35,11 @@ def main():
p_list = subparsers.add_parser("list", help="List all tools")
p_list.add_argument("--json", action="store_true", help="Output as JSON")
p_list.add_argument("--filter", help="Filter by name or description (substring match)")
p_list.add_argument("--limit", type=int, help="Return at most this many matching tools")
p_list.add_argument(
"--full", action="store_true",
help="Include complete definitions in JSON output (prompts and code)",
)
p_list.set_defaults(func=cmd_list)
# 'create' command
@ -81,8 +86,18 @@ def main():
# 'run-once' command — one-shot without creating a tool
p_run_once = subparsers.add_parser("run-once", help="Run a one-shot prompt through a provider without creating a tool")
p_run_once.add_argument("prompt", help="Prompt text (or '-' to read from stdin)")
p_run_once.add_argument("-p", "--provider", default="opencode-pickle", help="Provider to use")
p_run_once.add_argument(
"prompt",
help=(
"Prompt or template; piped/file input replaces {input}, or is "
"appended when the placeholder is absent ('-' reads stdin as the whole prompt)"
),
)
p_run_once.add_argument("-i", "--input", help="Read input data from this file (or '-' for stdin)")
p_run_once.add_argument(
"-p", "--provider",
help="Provider to use (default: configured provider, then first available provider)",
)
p_run_once.add_argument("--max-tokens", type=int, help="Max output tokens")
p_run_once.add_argument("--timeout", type=int, default=60, help="Timeout in seconds")
p_run_once.set_defaults(func=cmd_run_once)

185
src/cmdforge/cli/tool_commands.py
View File

@ -13,38 +13,50 @@ from ..gui import run_gui
def cmd_list(args):
"""List all tools."""
import json
import sys
tools = list_tools()
filter_text = (getattr(args, "filter", "") or "").casefold()
limit = getattr(args, "limit", None)
if getattr(args, "full", False) and not getattr(args, "json", False):
print("Error: --full requires --json.", file=sys.stderr)
return 1
if limit is not None and limit < 1:
print("Error: --limit must be at least 1.", file=sys.stderr)
return 1
filter_text = getattr(args, "filter", "") or ""
matches = []
for name in sorted(list_tools()):
tool = load_tool(name)
if not tool:
continue
if filter_text and (
filter_text not in name.casefold()
and filter_text not in (tool.description or "").casefold()
):
continue
matches.append((name, tool))
if limit is not None and len(matches) >= limit:
break
if getattr(args, "json", False):
result = []
for name in sorted(tools):
tool = load_tool(name)
if not tool:
continue
if filter_text and filter_text.lower() not in name.lower() and filter_text.lower() not in (tool.description or "").lower():
continue
result.append(tool.to_dict())
full = getattr(args, "full", False)
result = [
_full_tool_entry(name, tool) if full else _catalog_tool_entry(name, tool)
for name, tool in matches
]
json.dump(result, sys.stdout, indent=2)
print()
return 0
if not tools:
print("No tools found.")
print("Create your first tool with: cmdforge ui")
if not matches:
if filter_text:
print(f"No tools match filter '{getattr(args, 'filter', '')}'.")
else:
print("No tools found.")
print("Create your first tool with: cmdforge ui")
return 0
print(f"Available tools ({len(tools)}):\n")
for name in tools:
tool = load_tool(name)
if not tool:
continue
if filter_text and filter_text.lower() not in name.lower() and filter_text.lower() not in (tool.description or "").lower():
continue
print(f"Available tools ({len(matches)}):\n")
for name, tool in matches:
# Show source indicator for imported tools
source_marker = ""
if tool.source and tool.source.type == "imported":
@ -77,23 +89,71 @@ def cmd_list(args):
return 0
def _catalog_tool_entry(name, tool) -> dict:
"""Return the stable, compact discovery contract intended for agents."""
entry = {
"name": name,
"description": tool.description or "",
"version": tool.version or "",
"category": tool.category or "Other",
"arguments": [argument.to_dict() for argument in tool.arguments],
"input_schema": tool.input_schema,
"output_schema": tool.output_schema,
"deprecated": bool(tool.deprecated),
}
if tool.replacement:
entry["replacement"] = tool.replacement
return entry
def _full_tool_entry(name, tool) -> dict:
"""Return an explicitly requested complete definition."""
entry = tool.to_dict()
# list_tools() owns the canonical local/qualified reference. A config's
# internal name may be only the short name.
entry["name"] = name
return entry
def cmd_run_once(args):
"""Run a one-shot prompt through a provider without creating a tool."""
import sys
from ..providers import call_provider
prompt = args.prompt
if prompt == "-":
prompt = sys.stdin.read()
timeout = getattr(args, "timeout", 60)
max_tokens = getattr(args, "max_tokens", None)
if timeout < 1 or timeout > 3600:
print("Error: --timeout must be between 1 and 3600 seconds.", file=sys.stderr)
return 1
if max_tokens is not None and not 1 <= max_tokens <= 1_000_000:
print("Error: --max-tokens must be between 1 and 1000000.", file=sys.stderr)
return 1
try:
prompt = _build_one_shot_prompt(
args.prompt, getattr(args, "input", None), sys.stdin
)
except (OSError, ValueError) as exc:
print(f"Error reading input: {exc}", file=sys.stderr)
return 1
if not prompt.strip():
print("Error: No prompt provided.", file=sys.stderr)
return 1
provider = args.provider or "opencode-pickle"
provider = _resolve_one_shot_provider(getattr(args, "provider", None))
if not provider:
print(
"Error: No usable provider found. Pass --provider, set "
"'cmdforge config set default_provider NAME', or run "
"'cmdforge providers install'.",
file=sys.stderr,
)
return 1
print(f"[run-once] Provider: {provider}", file=sys.stderr)
result = call_provider(provider, prompt, timeout=args.timeout, max_tokens=args.max_tokens)
result = call_provider(
provider, prompt, timeout=timeout, max_tokens=max_tokens
)
if not result.success:
print(f"Error: {result.error}", file=sys.stderr)
@ -103,6 +163,75 @@ def cmd_run_once(args):
return 0
def _build_one_shot_prompt(template: str, input_path, stdin) -> str:
"""Combine a one-shot instruction with explicit or piped input."""
if template == "-":
if input_path and input_path != "-":
raise ValueError("prompt '-' cannot be combined with --input")
return stdin.read()
input_text = None
if input_path:
input_text = (
stdin.read()
if input_path == "-"
else Path(input_path).read_text(encoding="utf-8")
)
elif not stdin.isatty():
input_text = stdin.read()
if not input_text:
return template
if "{input}" in template:
return template.replace("{input}", input_text)
return f"{template.rstrip()}\n\n{input_text}"
def _resolve_one_shot_provider(explicit: str = None):
"""Honor the configured default, otherwise choose an available provider."""
if explicit:
return explicit
from ..config import load_config
from ..providers import PRESET_CHAINS, load_providers
configured = load_config().default_provider
if configured:
return configured
providers = {provider.name: provider for provider in load_providers()}
ordered_names = list(PRESET_CHAINS["balanced"][:-1])
ordered_names.extend(
name for name in providers if name not in ordered_names and name != "mock"
)
for name in ordered_names:
provider = providers.get(name)
if provider and _provider_is_available(provider):
return name
return None
def _provider_is_available(provider) -> bool:
"""Perform a non-invasive availability check without contacting a model."""
import os
import shlex
import shutil
if provider.type == "api":
key_name = provider.api_key_env or (
f"{provider.name.upper().replace('-', '_')}_API_KEY"
)
return bool(os.environ.get(key_name))
try:
parts = shlex.split(os.path.expandvars(provider.command))
except ValueError:
return False
if not parts:
return False
executable = Path(os.path.expanduser(parts[0]))
return bool(shutil.which(str(executable)) or executable.is_file())
def cmd_create(args):
"""Create a new tool (basic CLI creation - use 'ui' for full builder)."""
name = args.name

154
tests/test_cli.py
View File

@ -1,5 +1,6 @@
"""Tests for CLI commands."""
import json
import tempfile
from pathlib import Path
from types import SimpleNamespace
@ -67,6 +68,159 @@ class TestListCommand:
assert 'test-tool' in captured.out
assert 'another-tool' in captured.out
def test_json_is_compact_catalog_unless_full_requested(
self, temp_tools_dir, capsys
):
from cmdforge.tool import save_tool
save_tool(Tool(
name="catalog-tool",
description="Catalog entry",
input_schema={"type": "string"},
output_schema={"type": "string"},
steps=[PromptStep(
prompt="Private implementation {input}",
provider="mock",
output_var="result",
plain_text=True,
)],
output="{result}",
))
with patch('sys.argv', ['cmdforge', 'list', '--json']):
assert main() == 0
compact = json.loads(capsys.readouterr().out)
assert compact[0]["name"] == "catalog-tool"
assert compact[0]["input_schema"] == {"type": "string"}
assert "steps" not in compact[0]
with patch('sys.argv', ['cmdforge', 'list', '--json', '--full']):
assert main() == 0
full = json.loads(capsys.readouterr().out)
assert full[0]["steps"][0]["prompt"] == "Private implementation {input}"
def test_filter_reports_filtered_count_and_empty_result(
self, temp_tools_dir, capsys
):
from cmdforge.tool import save_tool
save_tool(Tool(name="alpha", description="First"))
save_tool(Tool(name="beta", description="Second"))
with patch('sys.argv', ['cmdforge', 'list', '--filter', 'alpha']):
assert main() == 0
output = capsys.readouterr().out
assert "Available tools (1)" in output
assert "alpha" in output
assert "beta" not in output
with patch('sys.argv', ['cmdforge', 'list', '--filter', 'missing']):
assert main() == 0
assert "No tools match filter 'missing'" in capsys.readouterr().out
def test_json_limit_and_invalid_limit(self, temp_tools_dir, capsys):
from cmdforge.tool import save_tool
save_tool(Tool(name="alpha"))
save_tool(Tool(name="beta"))
with patch('sys.argv', ['cmdforge', 'list', '--json', '--limit', '1']):
assert main() == 0
assert len(json.loads(capsys.readouterr().out)) == 1
with patch('sys.argv', ['cmdforge', 'list', '--limit', '0']):
assert main() == 1
assert "at least 1" in capsys.readouterr().err
with patch('sys.argv', ['cmdforge', 'list', '--full']):
assert main() == 1
assert "--full requires --json" in capsys.readouterr().err
class TestRunOnceCommand:
def test_combines_instruction_with_piped_input(self, capsys):
result = SimpleNamespace(success=True, text="summary", error=None)
with (
patch('sys.argv', [
'cmdforge', 'run-once', 'Summarize:\n{input}',
'--provider', 'mock',
]),
patch('sys.stdin', StringIO('document body')),
patch('cmdforge.providers.call_provider', return_value=result) as call_provider,
):
assert main() == 0
assert call_provider.call_args.args[:2] == (
'mock', 'Summarize:\ndocument body'
)
assert capsys.readouterr().out == "summary\n"
def test_appends_piped_input_when_template_has_no_placeholder(self):
result = SimpleNamespace(success=True, text="ok", error=None)
with (
patch('sys.argv', [
'cmdforge', 'run-once', 'Summarize this', '--provider', 'mock',
]),
patch('sys.stdin', StringIO('document body')),
patch('cmdforge.providers.call_provider', return_value=result) as call_provider,
):
assert main() == 0
assert call_provider.call_args.args[1] == "Summarize this\n\ndocument body"
def test_uses_configured_default_provider(self):
result = SimpleNamespace(success=True, text="ok", error=None)
config = SimpleNamespace(default_provider="configured-provider")
stdin = MagicMock()
stdin.isatty.return_value = True
with (
patch('sys.argv', ['cmdforge', 'run-once', 'hello']),
patch('sys.stdin', stdin),
patch('cmdforge.config.load_config', return_value=config),
patch('cmdforge.providers.call_provider', return_value=result) as call_provider,
):
assert main() == 0
assert call_provider.call_args.args[0] == "configured-provider"
def test_selects_first_available_balanced_provider(self):
from cmdforge.cli.tool_commands import _resolve_one_shot_provider
from cmdforge.providers import Provider
config = SimpleNamespace(default_provider=None)
providers = [
Provider("opencode-pickle", "missing-opencode"),
Provider("codex", "codex exec -"),
]
with (
patch('cmdforge.config.load_config', return_value=config),
patch('cmdforge.providers.load_providers', return_value=providers),
patch(
'cmdforge.cli.tool_commands._provider_is_available',
side_effect=lambda provider: provider.name == "codex",
),
):
assert _resolve_one_shot_provider() == "codex"
@pytest.mark.parametrize(
"arguments,message",
[
(["--timeout", "0"], "--timeout"),
(["--max-tokens", "-1"], "--max-tokens"),
],
)
def test_rejects_invalid_limits_before_provider_call(
self, arguments, message, capsys
):
with (
patch('sys.argv', [
'cmdforge', 'run-once', 'hello', '--provider', 'mock', *arguments,
]),
patch('sys.stdin', StringIO()),
patch('cmdforge.providers.call_provider') as call_provider,
):
assert main() == 1
call_provider.assert_not_called()
assert message in capsys.readouterr().err
class TestCreateCommand:
"""Tests for 'cmdforge create' command."""