symfonic.agent.cli.main

main

Symfonic agent CLI.

Commands

chat -- Run a single query against the agent. migrate -- Placeholder for future HMS data migration.

Usage

python -m symfonic.agent.cli.main chat "Hello" --tenant-id t1 symfonic chat "Hello" --tenant-id t1 symfonic migrate old://db new://db

chat

chat(query: Annotated[str, Argument(help='Query to send to the agent.')], tenant_id: Annotated[str, Option('--tenant-id', help='Tenant identifier (required).')], dsn: Annotated[str | None, Option('--dsn', help='Postgres DSN for persistent memory.')] = None, model: Annotated[str, Option('--model', help='Model provider name (default: mock).')] = 'mock', enable_hms: Annotated[bool, Option('--enable-hms', help='Enable HMS-aware system prompt.')] = False, domain: Annotated[str, Option('--domain', help="Domain template name. Use 'default' for generic agent. Custom domains should be injected programmatically.")] = 'default') -> None

Run a single query against the Symfonic agent.

This command is a mapper: it decodes the arguments, hands them to :func:symfonic.cli.chat.run_chat, and formats the result. Which runtime executes the turn -- the public symfonic.Agent facade for a plain invocation, SymfonicAgent when HMS memory, a named domain, or a DSN is asked for -- is decided there, where it can be tested without a CLI runner.

Source code in src/symfonic/agent/cli/main.py

@app.command()
def chat(
    query: Annotated[str, typer.Argument(help="Query to send to the agent.")],
    tenant_id: Annotated[
        str, typer.Option("--tenant-id", help="Tenant identifier (required).")
    ],
    dsn: Annotated[
        str | None, typer.Option("--dsn", help="Postgres DSN for persistent memory.")
    ] = None,
    model: Annotated[
        str, typer.Option("--model", help="Model provider name (default: mock).")
    ] = "mock",
    enable_hms: Annotated[
        bool, typer.Option("--enable-hms", help="Enable HMS-aware system prompt.")
    ] = False,
    domain: Annotated[
        str,
        typer.Option(
            "--domain",
            help="Domain template name. Use 'default' for generic agent. "
            "Custom domains should be injected programmatically.",
        ),
    ] = "default",
) -> None:
    """Run a single query against the Symfonic agent.

    This command is a mapper: it decodes the arguments, hands them to
    :func:`symfonic.cli.chat.run_chat`, and formats the result. Which runtime
    executes the turn -- the public ``symfonic.Agent`` facade for a plain
    invocation, ``SymfonicAgent`` when HMS memory, a named domain, or a DSN is
    asked for -- is decided there, where it can be tested without a CLI runner.
    """
    if dsn:
        typer.echo(
            "Warning: --dsn is not yet wired to Postgres backends. Using in-memory.",
            err=True,
        )

    try:
        provider = build_provider(model, query)
    except ChatError as exc:
        typer.echo(str(exc), err=True)
        raise SystemExit(1) from exc

    request = ChatRequest(
        query=query,
        tenant_id=tenant_id,
        dsn=dsn,
        model=model,
        enable_hms=bool(enable_hms),
        domain=domain,
    )

    if enable_hms:
        print("Thinking (HMS enabled)...", file=sys.stderr)

    outcome = asyncio.run(run_chat(request, provider))

    typer.echo(outcome.text or "(no response)")
    in_tok = len(query) // 3
    out_tok = len(outcome.text) // 3
    sys_tok = outcome.system_prompt_tokens
    total_tok = in_tok + sys_tok + out_tok
    print(
        f"tokens=~{total_tok} (in=~{in_tok} +system=~{sys_tok} out=~{out_tok})  "
        f"memory={outcome.memory_entries_used}  "
        f"duration_ms={outcome.duration_ms:.1f}",
        file=sys.stderr,
    )

examples_add

examples_add(name: Annotated[str, Argument(help="Example to copy (see 'symfonic examples list').")], dest: Annotated[Path, Option('--dir', '-d', help='Destination directory (default: cwd).')] = Path('.'), force: Annotated[bool, Option('--force', help='Overwrite an existing target directory.')] = False) -> None

Copy a curated example into <dir>/<name>/ and print how to run it.

Source code in src/symfonic/agent/cli/main.py

@examples_app.command("add")
def examples_add(
    name: Annotated[
        str, typer.Argument(help="Example to copy (see 'symfonic examples list').")
    ],
    dest: Annotated[
        Path,
        typer.Option("--dir", "-d", help="Destination directory (default: cwd)."),
    ] = Path("."),
    force: Annotated[
        bool, typer.Option("--force", help="Overwrite an existing target directory.")
    ] = False,
) -> None:
    """Copy a curated example into ``<dir>/<name>/`` and print how to run it."""
    from symfonic.cli.examples import CURATED, ExamplesError, add_example

    try:
        target = add_example(name, dest, force=force)
    except ExamplesError as exc:
        typer.echo(f"error: {exc}", err=True)
        raise SystemExit(1) from exc

    spec = CURATED[name]
    typer.echo(f"Copied '{name}' -> {target}")
    if spec.requires:
        typer.echo(f"\nRequires: {spec.requires}")
    typer.echo("\nRun it:")
    typer.echo(f"  {spec.run}")

examples_list

examples_list() -> None

List the curated examples available to copy.

Source code in src/symfonic/agent/cli/main.py

@examples_app.command("list")
def examples_list() -> None:
    """List the curated examples available to copy."""
    from symfonic.cli.examples import available_examples

    typer.echo("Available examples (copy with: symfonic examples add <name>):\n")
    width = max(len(s.name) for s in available_examples())
    for spec in available_examples():
        line = f"  {spec.name.ljust(width)}  {spec.summary}"
        if spec.requires:
            line += f"  [requires: {spec.requires}]"
        typer.echo(line)

guide

guide() -> None

Print the agent quick reference (decision tree, imports, footguns).

The reference (AGENTS.md) ships inside the wheel, so this works offline with no repo checkout -- the canonical entry point for an agent or person figuring out how to build with symfonic-core from an installed package.

Source code in src/symfonic/agent/cli/main.py

@app.command()
def guide() -> None:
    """Print the agent quick reference (decision tree, imports, footguns).

    The reference (``AGENTS.md``) ships inside the wheel, so this works offline
    with no repo checkout -- the canonical entry point for an agent or person
    figuring out how to build with symfonic-core from an installed package.
    """
    from importlib.resources import files

    try:
        text = files("symfonic.agent").joinpath("AGENTS.md").read_text(
            encoding="utf-8",
        )
    except (FileNotFoundError, ModuleNotFoundError):  # pragma: no cover
        typer.echo(
            "AGENTS.md is not bundled in this install. See "
            "https://core.symfonic.dev",
            err=True,
        )
        raise SystemExit(1) from None
    typer.echo(text)

migrate

migrate(source_uri: Annotated[str, Argument(help='Source data URI.')], target_uri: Annotated[str, Argument(help='Target data URI.')]) -> None

Migrate HMS data between storage backends (stub).

Source code in src/symfonic/agent/cli/main.py

@app.command()
def migrate(
    source_uri: Annotated[str, typer.Argument(help="Source data URI.")],
    target_uri: Annotated[str, typer.Argument(help="Target data URI.")],
) -> None:
    """Migrate HMS data between storage backends (stub)."""
    typer.echo(f"Migration from {source_uri} to {target_uri} is not yet implemented")
    raise SystemExit(0)