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', retrieval_min_relevance: Annotated[float, Option('--retrieval-min-relevance', min=0.0, max=1.0, help='Native HMS relevance floor (default: 0.0).')] = 0.0) -> 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 -- a bare or presented public Agent, or a native durable HMS composition -- is decided there.

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",
    retrieval_min_relevance: Annotated[
        float,
        typer.Option(
            "--retrieval-min-relevance", min=0.0, max=1.0,
            help="Native HMS relevance floor (default: 0.0).",
        ),
    ] = 0.0,
) -> 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 -- a bare or presented public ``Agent``, or a native
    durable HMS composition -- is decided there.
    """
    if enable_hms and not dsn:
        typer.echo("--enable-hms requires --dsn with a PostgreSQL database", err=True)
        raise SystemExit(1)
    if enable_hms and model == "mock":
        typer.echo(
            "--enable-hms requires a real model provider; use --model anthropic "
            "(the mock provider cannot extract durable memories)",
            err=True,
        )
        raise SystemExit(1)
    if not 0.0 <= retrieval_min_relevance <= 1.0:
        typer.echo("--retrieval-min-relevance must be between 0.0 and 1.0", err=True)
        raise SystemExit(1)
    if retrieval_min_relevance > 0.0 and not enable_hms:
        typer.echo("--retrieval-min-relevance requires --enable-hms", err=True)
        raise SystemExit(1)
    if dsn and not enable_hms:
        typer.echo(
            "Warning: --dsn is not yet wired to Postgres backends. "
            "No persistent store is configured.",
            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,
        retrieval_min_relevance=retrieval_min_relevance,
    )

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

    from symfonic.capabilities.memory import MemoryContractError

    try:
        outcome = asyncio.run(run_chat(request, provider))
    except ChatError as exc:
        typer.echo(str(exc), err=True)
        raise SystemExit(1) from None
    except MemoryContractError:
        typer.echo(
            "Invalid --tenant-id for HMS: use letters, digits, _, ., :, @, or -.",
            err=True,
        )
        raise SystemExit(1) from None
    except Exception as exc:
        if not enable_hms:
            raise
        # Driver exceptions can include the connection URL, including its
        # password. The CLI reports the failure class, never its message.
        typer.echo(
            f"HMS turn failed ({type(exc).__name__}). Check PostgreSQL "
            "connectivity/schema and provider configuration.",
            err=True,
        )
        raise SystemExit(1) from None

    typer.echo(outcome.text or "(no response)")
    in_tok = len(query) // 3
    out_tok = len(outcome.text) // 3
    if outcome.native_hms is not None:
        native = outcome.native_hms
        extraction = native.stages.get("memory.consolidation")
        lifecycle = native.stages.get("memory.lifecycle")
        print(
            f"input_tokens_est=~{in_tok} output_tokens_est=~{out_tok} "
            f"prompt_admitted_tokens_est=~{native.admitted_contribution_tokens} "
            f"memory_admitted={native.recalled_records_count} "
            f"memory_published={lifecycle.counts.get('published', 0) if lifecycle else 0} "
            f"extraction={extraction.outcome if extraction else 'missing'} "
            f"duration_ms={outcome.duration_ms:.1f}",
            file=sys.stderr,
        )
    else:
        sys_tok = outcome.system_prompt_tokens or 0
        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)