memory.artifact_tools

memory.artifact_tools

Artifact, pipeline, and experiment MCP tools.

This block manages DVC-tracked large files, S3 remotes, DVC pipelines, and experiment reproduction/rerun. It is graph-free — every tool delegates to memory.artifacts (and memory.jobs for cloud reruns), touching only DVC, S3, manifests, and the filesystem. Because it never opens the KGLite store, it is hosted in its own memory-artifacts MCP server process (see memory/artifacts_server.py) so it can be toggled independently of the core memory server without risking a concurrent-borrow panic on the shared cache.

Tools are defined at module level and registered onto a server via :func:register. They call the module-level _schedule_memory_commit hook (a no-op until register wires in the host server's committer) so that manifest/dvc.yaml writes get committed.

configure_artifact_remote

configure_artifact_remote(bucket: str, prefix: str = '', region: str = '', remote_name: str = DEFAULT_REMOTE_NAME, endpoint_url: str = '', force: bool = False, dry_run: bool = True) -> str

Configure the DVC S3 remote used for artifact push/pull.

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
prefix	str	Optional bucket prefix for this repo/project.	''
region	str	Optional AWS region.	''
remote_name	str	DVC remote name.	DEFAULT_REMOTE_NAME
endpoint_url	str	Optional S3-compatible endpoint URL.	''
force	bool	Replace an existing remote with the same name.	False
dry_run	bool	If true, return the commands without changing DVC config.	True

Source code in memory/artifact_tools.py

def configure_artifact_remote(
    bucket: str,
    prefix: str = "",
    region: str = "",
    remote_name: str = artifacts.DEFAULT_REMOTE_NAME,
    endpoint_url: str = "",
    force: bool = False,
    dry_run: bool = True,
) -> str:
    """Configure the DVC S3 remote used for artifact push/pull.

    Args:
        bucket: S3 bucket name.
        prefix: Optional bucket prefix for this repo/project.
        region: Optional AWS region.
        remote_name: DVC remote name.
        endpoint_url: Optional S3-compatible endpoint URL.
        force: Replace an existing remote with the same name.
        dry_run: If true, return the commands without changing DVC config.
    """
    try:
        config = artifacts.S3RemoteConfig(
            bucket=bucket,
            prefix=prefix,
            region=region or None,
            remote_name=remote_name or artifacts.DEFAULT_REMOTE_NAME,
            endpoint_url=endpoint_url or None,
        )
        results = artifacts.configure_s3_remote(config, dry_run=dry_run, force=force)
        return json.dumps({
            "remote": config.remote_name,
            "url": config.url,
            "dry_run": dry_run,
            "commands": [artifacts.command_result_to_dict(result) for result in results],
            "message": "No AWS connection is attempted by this tool.",
        })
    except Exception as exc:
        return _json_tool_error(exc)

add_artifact

add_artifact(paths: str, dry_run: bool = True) -> str

Register one or more repository-local paths with DVC.

Parameters:

Name	Type	Description	Default
paths	str	Comma/newline-separated paths or a JSON list of paths.	required
dry_run	bool	If true, return the DVC command without executing it.	True

Source code in memory/artifact_tools.py

def add_artifact(paths: str, dry_run: bool = True) -> str:
    """Register one or more repository-local paths with DVC.

    Args:
        paths: Comma/newline-separated paths or a JSON list of paths.
        dry_run: If true, return the DVC command without executing it.
    """
    try:
        parsed = _parse_artifact_paths(paths)
        result = artifacts.dvc_add(parsed, dry_run=dry_run)
        return json.dumps({
            "paths": parsed,
            "dry_run": dry_run,
            "result": artifacts.command_result_to_dict(result),
            "message": "DVC stores large artifact content; Git should store generated .dvc pointers.",
        })
    except Exception as exc:
        return _json_tool_error(exc)

push_artifact

push_artifact(targets: str = '', dry_run: bool = True) -> str

Push DVC-tracked artifacts to the configured remote.

Parameters:

Name	Type	Description	Default
targets	str	Optional comma/newline-separated target paths or JSON list.	''
dry_run	bool	If true, return the DVC command without contacting the remote.	True

Source code in memory/artifact_tools.py

def push_artifact(targets: str = "", dry_run: bool = True) -> str:
    """Push DVC-tracked artifacts to the configured remote.

    Args:
        targets: Optional comma/newline-separated target paths or JSON list.
        dry_run: If true, return the DVC command without contacting the remote.
    """
    try:
        parsed = _parse_artifact_paths(targets)
        result = artifacts.dvc_push(targets=parsed or None, dry_run=dry_run)
        return json.dumps({
            "targets": parsed,
            "dry_run": dry_run,
            "result": artifacts.command_result_to_dict(result),
        })
    except Exception as exc:
        return _json_tool_error(exc)

pull_artifact

pull_artifact(targets: str = '', dry_run: bool = True) -> str

Pull DVC-tracked artifacts from the configured remote.

Parameters:

Name	Type	Description	Default
targets	str	Optional comma/newline-separated target paths or JSON list.	''
dry_run	bool	If true, return the DVC command without contacting the remote.	True

Source code in memory/artifact_tools.py

def pull_artifact(targets: str = "", dry_run: bool = True) -> str:
    """Pull DVC-tracked artifacts from the configured remote.

    Args:
        targets: Optional comma/newline-separated target paths or JSON list.
        dry_run: If true, return the DVC command without contacting the remote.
    """
    try:
        parsed = _parse_artifact_paths(targets)
        result = artifacts.dvc_pull(targets=parsed or None, dry_run=dry_run)
        return json.dumps({
            "targets": parsed,
            "dry_run": dry_run,
            "result": artifacts.command_result_to_dict(result),
        })
    except Exception as exc:
        return _json_tool_error(exc)

verify_artifacts

verify_artifacts(paths: str = '', dry_run: bool = True) -> str

Verify local artifact/pointer presence and DVC status.

Parameters:

Name	Type	Description	Default
paths	str	Optional comma/newline-separated paths or JSON list.	''
dry_run	bool	If true, return the DVC status command without executing it.	True

Source code in memory/artifact_tools.py

def verify_artifacts(paths: str = "", dry_run: bool = True) -> str:
    """Verify local artifact/pointer presence and DVC status.

    Args:
        paths: Optional comma/newline-separated paths or JSON list.
        dry_run: If true, return the DVC status command without executing it.
    """
    try:
        parsed = _parse_artifact_paths(paths)
        result = artifacts.verify_artifacts(parsed or None, dry_run=dry_run)
        result["dry_run"] = dry_run
        return json.dumps(result)
    except Exception as exc:
        return _json_tool_error(exc)

list_artifacts

list_artifacts(include_manifests: bool = True) -> str

List DVC artifact pointers and stored experiment manifests.

Source code in memory/artifact_tools.py

def list_artifacts(include_manifests: bool = True) -> str:
    """List DVC artifact pointers and stored experiment manifests."""
    try:
        payload = {
            "artifacts": artifacts.list_artifacts(),
        }
        if include_manifests:
            payload["manifests"] = artifacts.list_manifests()
        return json.dumps(payload, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

resolve_artifact

resolve_artifact(identifier: str) -> str

Resolve an artifact path/id to a DVC pointer, manifest, or local path.

Source code in memory/artifact_tools.py

def resolve_artifact(identifier: str) -> str:
    """Resolve an artifact path/id to a DVC pointer, manifest, or local path."""
    try:
        return json.dumps(artifacts.resolve_artifact(identifier), default=str)
    except Exception as exc:
        return _json_tool_error(exc)

list_pipeline_stages

list_pipeline_stages() -> str

List DVC pipeline stages from dvc.yaml.

Source code in memory/artifact_tools.py

def list_pipeline_stages() -> str:
    """List DVC pipeline stages from dvc.yaml."""
    try:
        return json.dumps({"stages": artifacts.list_pipeline_stages()}, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

add_pipeline_stage

add_pipeline_stage(name: str, command: str, deps: str = '', outs: str = '', params: str = '', metrics: str = '', plots: str = '', wdir: str = '', force: bool = False, dry_run: bool = True) -> str

Add or update a DVC pipeline stage.

Parameters:

Name	Type	Description	Default
name	str	DVC stage name.	required
command	str	Stage command to execute.	required
deps	str	Comma/newline-separated dependency paths or JSON list.	''
outs	str	Comma/newline-separated output paths or JSON list.	''
params	str	Comma/newline-separated DVC params keys or JSON list.	''
metrics	str	Comma/newline-separated metrics paths or JSON list.	''
plots	str	Comma/newline-separated plot paths or JSON list.	''
wdir	str	Optional repo-local working directory for the stage.	''
force	bool	Pass --force when updating an existing stage.	False
dry_run	bool	If true, return the planned DVC command without writing dvc.yaml.	True

Source code in memory/artifact_tools.py

def add_pipeline_stage(
    name: str,
    command: str,
    deps: str = "",
    outs: str = "",
    params: str = "",
    metrics: str = "",
    plots: str = "",
    wdir: str = "",
    force: bool = False,
    dry_run: bool = True,
) -> str:
    """Add or update a DVC pipeline stage.

    Args:
        name: DVC stage name.
        command: Stage command to execute.
        deps: Comma/newline-separated dependency paths or JSON list.
        outs: Comma/newline-separated output paths or JSON list.
        params: Comma/newline-separated DVC params keys or JSON list.
        metrics: Comma/newline-separated metrics paths or JSON list.
        plots: Comma/newline-separated plot paths or JSON list.
        wdir: Optional repo-local working directory for the stage.
        force: Pass --force when updating an existing stage.
        dry_run: If true, return the planned DVC command without writing dvc.yaml.
    """
    try:
        result = artifacts.dvc_stage_add(
            name=name,
            command=command,
            deps=_parse_artifact_paths(deps) or None,
            outs=_parse_artifact_paths(outs) or None,
            params=_parse_artifact_paths(params) or None,
            metrics=_parse_artifact_paths(metrics) or None,
            plots=_parse_artifact_paths(plots) or None,
            wdir=wdir or None,
            force=force,
            dry_run=dry_run,
        )
        if not dry_run:
            _schedule_memory_commit("dvc.yaml")
        return json.dumps({
            "name": name,
            "dry_run": dry_run,
            "result": artifacts.command_result_to_dict(result),
        }, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

run_pipeline

run_pipeline(pipeline_id: str, targets: str = '', dry_run: bool = True, require_remote: bool = False, environment_json: str = '') -> str

Run or plan DVC pipeline reproduction and write a manifest.

Parameters:

Name	Type	Description	Default
pipeline_id	str	Stable id for the pipeline-run manifest.	required
targets	str	Optional comma/newline-separated DVC stage targets or JSON list.	''
dry_run	bool	If true, plan the DVC repro command without executing it.	True
require_remote	bool	If true, push DVC outputs before treating the run as portable.	False
environment_json	str	Optional JSON object to override the environment summary.	''

Source code in memory/artifact_tools.py

def run_pipeline(
    pipeline_id: str,
    targets: str = "",
    dry_run: bool = True,
    require_remote: bool = False,
    environment_json: str = "",
) -> str:
    """Run or plan DVC pipeline reproduction and write a manifest.

    Args:
        pipeline_id: Stable id for the pipeline-run manifest.
        targets: Optional comma/newline-separated DVC stage targets or JSON list.
        dry_run: If true, plan the DVC repro command without executing it.
        require_remote: If true, push DVC outputs before treating the run as portable.
        environment_json: Optional JSON object to override the environment summary.
    """
    try:
        parsed_targets = _parse_artifact_paths(targets)
        environment = _parse_json_object(environment_json, "environment_json")
        result = artifacts.run_pipeline(
            pipeline_id=pipeline_id,
            targets=parsed_targets,
            dry_run=dry_run,
            require_remote=require_remote,
            environment=environment or None,
        )
        _schedule_memory_commit(result["manifest_path"])
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

run_experiment

run_experiment(experiment_id: str, command: str, artifacts_json: str = '', cwd: str = '', execute: bool = False, dry_run: bool = True, require_remote: bool = False, include_dirty_diff: bool = False, dirty_note: str = '', environment_json: str = '') -> str

Create an experiment manifest and optionally execute the command.

Parameters:

Name	Type	Description	Default
experiment_id	str	Stable id for the manifest filename.	required
command	str	Command needed to reproduce the experiment.	required
artifacts_json	str	JSON list of artifact records, JSON list of paths, or comma-separated paths.	''
cwd	str	Optional repo-local working directory for the command.	''
execute	bool	If true and dry_run is false, execute the command.	False
dry_run	bool	Default true; captures provenance without running the command.	True
require_remote	bool	If true, push DVC artifacts before treating the manifest as portable.	False
include_dirty_diff	bool	Include a git diff stat in the manifest.	False
dirty_note	str	Optional human note about dirty working-tree state.	''
environment_json	str	Optional JSON object to override the environment summary.	''

Source code in memory/artifact_tools.py

def run_experiment(
    experiment_id: str,
    command: str,
    artifacts_json: str = "",
    cwd: str = "",
    execute: bool = False,
    dry_run: bool = True,
    require_remote: bool = False,
    include_dirty_diff: bool = False,
    dirty_note: str = "",
    environment_json: str = "",
) -> str:
    """Create an experiment manifest and optionally execute the command.

    Args:
        experiment_id: Stable id for the manifest filename.
        command: Command needed to reproduce the experiment.
        artifacts_json: JSON list of artifact records, JSON list of paths, or comma-separated paths.
        cwd: Optional repo-local working directory for the command.
        execute: If true and dry_run is false, execute the command.
        dry_run: Default true; captures provenance without running the command.
        require_remote: If true, push DVC artifacts before treating the manifest as portable.
        include_dirty_diff: Include a git diff stat in the manifest.
        dirty_note: Optional human note about dirty working-tree state.
        environment_json: Optional JSON object to override the environment summary.
    """
    try:
        parsed_artifacts = _parse_artifact_records(artifacts_json)
        environment = _parse_json_object(environment_json, "environment_json")
        result = artifacts.run_experiment(
            experiment_id=experiment_id,
            command=command,
            artifacts=parsed_artifacts,
            cwd=cwd or None,
            execute=execute,
            dry_run=dry_run,
            require_remote=require_remote,
            include_dirty_diff=include_dirty_diff,
            dirty_note=dirty_note,
            environment=environment or None,
        )
        _schedule_memory_commit(result["manifest_path"])
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

push_manifest_artifacts

push_manifest_artifacts(experiment_id: str, dry_run: bool = False) -> str

Push all DVC artifacts referenced by a manifest and update remote state.

Parameters:

Name	Type	Description	Default
experiment_id	str	Manifest id to load.	required
dry_run	bool	If true, return planned push semantics without remote I/O.	False

Source code in memory/artifact_tools.py

def push_manifest_artifacts(
    experiment_id: str,
    dry_run: bool = False,
) -> str:
    """Push all DVC artifacts referenced by a manifest and update remote state.

    Args:
        experiment_id: Manifest id to load.
        dry_run: If true, return planned push semantics without remote I/O.
    """
    try:
        result = artifacts.push_manifest_artifacts(experiment_id, dry_run=dry_run)
        manifest_path = result.get("manifest_path")
        if not manifest_path and isinstance(result.get("manifest"), dict):
            manifest_id = str(result["manifest"].get("id", experiment_id))
            manifest_path = str(artifacts.artifact_manifest_path(manifest_id).relative_to(Path.cwd()).as_posix())
        if manifest_path and not dry_run:
            _schedule_memory_commit(manifest_path)
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

verify_manifest

verify_manifest(experiment_id: str, dry_run: bool = False) -> str

Verify local artifact pointers/files and report manifest remote state.

Parameters:

Name	Type	Description	Default
experiment_id	str	Manifest id to load.	required
dry_run	bool	If true, return planned DVC status without executing it.	False

Source code in memory/artifact_tools.py

def verify_manifest(
    experiment_id: str,
    dry_run: bool = False,
) -> str:
    """Verify local artifact pointers/files and report manifest remote state.

    Args:
        experiment_id: Manifest id to load.
        dry_run: If true, return planned DVC status without executing it.
    """
    try:
        result = artifacts.verify_manifest(experiment_id, dry_run=dry_run)
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

export_experiment_catalog

export_experiment_catalog(experiment_id: str, catalog_prefix: str = '', copy_small_artifacts: bool = True, max_artifact_bytes: int = 1000000, dry_run: bool = True) -> str

Export a human-readable S3 catalog beside DVC's hash-addressed cache.

Parameters:

Name	Type	Description	Default
experiment_id	str	Manifest id to export.	required
catalog_prefix	str	Optional explicit s3:// prefix; defaults to /experiments/.	''
copy_small_artifacts	bool	Copy small non-input artifacts, such as metrics JSON, into the catalog.	True
max_artifact_bytes	int	Maximum artifact size to copy into the readable catalog.	1000000
dry_run	bool	If true, return planned AWS uploads without writing local catalog files or uploading.	True

Source code in memory/artifact_tools.py

def export_experiment_catalog(
    experiment_id: str,
    catalog_prefix: str = "",
    copy_small_artifacts: bool = True,
    max_artifact_bytes: int = 1000000,
    dry_run: bool = True,
) -> str:
    """Export a human-readable S3 catalog beside DVC's hash-addressed cache.

    Args:
        experiment_id: Manifest id to export.
        catalog_prefix: Optional explicit s3:// prefix; defaults to <DVC remote>/experiments/<id>.
        copy_small_artifacts: Copy small non-input artifacts, such as metrics JSON, into the catalog.
        max_artifact_bytes: Maximum artifact size to copy into the readable catalog.
        dry_run: If true, return planned AWS uploads without writing local catalog files or uploading.
    """
    try:
        result = artifacts.export_experiment_catalog(
            experiment_id=experiment_id,
            catalog_prefix=catalog_prefix,
            copy_small_artifacts=copy_small_artifacts,
            max_artifact_bytes=max_artifact_bytes,
            dry_run=dry_run,
        )
        if not dry_run:
            _schedule_memory_commit(result["local_dir"])
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

reproduce_experiment

reproduce_experiment(experiment_id: str, pull: bool = False, execute: bool = False, dry_run: bool = True) -> str

Prepare or execute reproduction steps from a stored manifest.

Parameters:

Name	Type	Description	Default
experiment_id	str	Manifest id to load.	required
pull	bool	Pull DVC artifacts listed in the manifest before running.	False
execute	bool	If true and dry_run is false, execute the manifest command.	False
dry_run	bool	Default true; returns planned DVC/command actions without remote I/O.	True

Source code in memory/artifact_tools.py

def reproduce_experiment(
    experiment_id: str,
    pull: bool = False,
    execute: bool = False,
    dry_run: bool = True,
) -> str:
    """Prepare or execute reproduction steps from a stored manifest.

    Args:
        experiment_id: Manifest id to load.
        pull: Pull DVC artifacts listed in the manifest before running.
        execute: If true and dry_run is false, execute the manifest command.
        dry_run: Default true; returns planned DVC/command actions without remote I/O.
    """
    try:
        result = artifacts.reproduce_experiment(
            experiment_id=experiment_id,
            pull=pull,
            execute=execute,
            dry_run=dry_run,
        )
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

rerun_experiment

rerun_experiment(experiment_id: str, where: str = 'cloud', executor: str = '', dry_run: bool = True) -> str

Rerun an experiment from its manifest in a clean sandbox.

Cloud mode (default) enqueues a job to the S3 queue for the AWS runner and returns a job_id to poll with get_rerun_job. Local mode runs the sandbox rerun on this machine (git worktree at the original commit + dvc pull + command + new dated run manifest with a replication check).

Parameters:

Name	Type	Description	Default
experiment_id	str	Manifest id of the run to rerun.	required
where	str	"cloud" to enqueue for the runner, "local" to run here.	'cloud'
executor	str	Optional executor label recorded in the manifest/job.	''
dry_run	bool	Default true; cloud mode returns the job spec without enqueueing, local mode returns the sandbox plan without executing.	True

Source code in memory/artifact_tools.py

def rerun_experiment(
    experiment_id: str,
    where: str = "cloud",
    executor: str = "",
    dry_run: bool = True,
) -> str:
    """Rerun an experiment from its manifest in a clean sandbox.

    Cloud mode (default) enqueues a job to the S3 queue for the AWS runner and
    returns a job_id to poll with get_rerun_job. Local mode runs the sandbox
    rerun on this machine (git worktree at the original commit + dvc pull +
    command + new dated run manifest with a replication check).

    Args:
        experiment_id: Manifest id of the run to rerun.
        where: "cloud" to enqueue for the runner, "local" to run here.
        executor: Optional executor label recorded in the manifest/job.
        dry_run: Default true; cloud mode returns the job spec without
            enqueueing, local mode returns the sandbox plan without executing.
    """
    try:
        if where == "cloud":
            from memory import jobs

            # Resolve failed-run ids back to the nearest rerunnable manifest.
            manifest = artifacts.resolve_rerun_target(experiment_id)
            job = jobs.build_rerun_job(manifest, requested_by="mcp", executor=executor or "aws-runner")
            if dry_run:
                return json.dumps({"dry_run": True, "job": job}, default=str)
            queue = jobs.default_queue()
            queue.enqueue(job)
            return json.dumps({
                "dry_run": False,
                "job_id": job["job_id"],
                "state": "pending",
                "queue": queue.base,
            }, default=str)
        result = artifacts.rerun_experiment(
            experiment_id,
            executor=executor or "local",
            dry_run=dry_run,
        )
        if not dry_run and result.get("manifest_path"):
            _schedule_memory_commit(result["manifest_path"])
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

get_rerun_job

get_rerun_job(job_id: str, tail_chars: int = 8000) -> str

Get the state and log tail of a queued/running/finished rerun job.

Parameters:

Name	Type	Description	Default
job_id	str	Job id returned by rerun_experiment(where="cloud").	required
tail_chars	int	Maximum number of trailing log characters to return.	8000

Source code in memory/artifact_tools.py

def get_rerun_job(job_id: str, tail_chars: int = 8000) -> str:
    """Get the state and log tail of a queued/running/finished rerun job.

    Args:
        job_id: Job id returned by rerun_experiment(where="cloud").
        tail_chars: Maximum number of trailing log characters to return.
    """
    try:
        from memory import jobs

        queue = jobs.default_queue()
        found = queue.get_job(job_id)
        if found is None:
            return _err(f"Job not found: {job_id}", "NotFound")
        state, job = found
        return json.dumps({
            "job_id": job_id,
            "state": state,
            "job": job,
            "log": queue.read_log(job_id, tail_chars=tail_chars),
        }, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

backup_artifact

backup_artifact(path: str, dry_run: bool = False) -> str

One-step backup of a local file: DVC add + push + verify.

The file stays where it is; this tracks it with DVC, ships a copy to the S3 remote, and returns the pointer/md5 so the path is restorable on any machine via dvc pull.

Parameters:

Name	Type	Description	Default
path	str	Repo-local path of the file or directory to back up.	required
dry_run	bool	If true, return planned add/push without remote I/O.	False

Source code in memory/artifact_tools.py

def backup_artifact(path: str, dry_run: bool = False) -> str:
    """One-step backup of a local file: DVC add + push + verify.

    The file stays where it is; this tracks it with DVC, ships a copy to the
    S3 remote, and returns the pointer/md5 so the path is restorable on any
    machine via dvc pull.

    Args:
        path: Repo-local path of the file or directory to back up.
        dry_run: If true, return planned add/push without remote I/O.
    """
    try:
        result = artifacts.backup_artifact(path, dry_run=dry_run)
        if not dry_run and result.get("pointer"):
            _schedule_memory_commit(result["pointer"])
        return json.dumps(result, default=str)
    except Exception as exc:
        return _json_tool_error(exc)

artifact

artifact(action: Literal['configure_remote', 'add', 'push', 'pull', 'verify', 'list', 'resolve', 'backup'], paths: str = '', targets: str = '', identifier: str = '', path: str = '', include_manifests: bool = True, bucket: str = '', prefix: str = '', region: str = '', remote_name: str = '', endpoint_url: str = '', force: bool = False, dry_run: bool | None = None) -> str

Use this when you need to manage a large/regenerable file via DVC: register, ship, retrieve, verify, resolve, or back it up.

Actions — name(required, optional?): configure_remote(bucket, prefix?, region?, remote_name?, endpoint_url?, force?, dry_run?): configure the DVC S3 remote. add(paths, dry_run?): register repo-local paths with DVC. push(targets?, dry_run?): push DVC-tracked artifacts to the remote. pull(targets?, dry_run?): pull DVC-tracked artifacts from the remote. verify(paths?, dry_run?): verify local pointer/file presence + DVC status. list(include_manifests?): list DVC pointers and stored manifests. resolve(identifier): resolve an identifier to a pointer/manifest/local path. backup(path, dry_run?): one-step DVC add + push + verify of a local path.

dry_run defaults to true for every action except backup (false). All return JSON.

Source code in memory/artifact_tools.py

def artifact(
    action: Literal[
        "configure_remote", "add", "push", "pull", "verify", "list", "resolve", "backup"
    ],
    paths: str = "",
    targets: str = "",
    identifier: str = "",
    path: str = "",
    include_manifests: bool = True,
    bucket: str = "",
    prefix: str = "",
    region: str = "",
    remote_name: str = "",
    endpoint_url: str = "",
    force: bool = False,
    dry_run: bool | None = None,
) -> str:
    """Use this when you need to manage a large/regenerable file via DVC: register, ship, retrieve, verify, resolve, or back it up.

    Actions — name(required, optional?):
        configure_remote(bucket, prefix?, region?, remote_name?, endpoint_url?, force?, dry_run?):
            configure the DVC S3 remote.
        add(paths, dry_run?): register repo-local paths with DVC.
        push(targets?, dry_run?): push DVC-tracked artifacts to the remote.
        pull(targets?, dry_run?): pull DVC-tracked artifacts from the remote.
        verify(paths?, dry_run?): verify local pointer/file presence + DVC status.
        list(include_manifests?): list DVC pointers and stored manifests.
        resolve(identifier): resolve an identifier to a pointer/manifest/local path.
        backup(path, dry_run?): one-step DVC add + push + verify of a local path.

    ``dry_run`` defaults to true for every action except ``backup`` (false).
    All return JSON.
    """
    if action == "configure_remote":
        return configure_artifact_remote(
            bucket=bucket, prefix=prefix, region=region, remote_name=remote_name,
            endpoint_url=endpoint_url, force=force,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "add":
        return add_artifact(paths=paths, dry_run=(True if dry_run is None else dry_run))
    if action == "push":
        return push_artifact(targets=targets, dry_run=(True if dry_run is None else dry_run))
    if action == "pull":
        return pull_artifact(targets=targets, dry_run=(True if dry_run is None else dry_run))
    if action == "verify":
        return verify_artifacts(paths=paths, dry_run=(True if dry_run is None else dry_run))
    if action == "list":
        return list_artifacts(include_manifests=include_manifests)
    if action == "resolve":
        return resolve_artifact(identifier=identifier)
    if action == "backup":
        return backup_artifact(path=path, dry_run=(False if dry_run is None else dry_run))
    return _bad_action(artifact, action)

pipeline

pipeline(action: Literal['list_stages', 'add_stage', 'run'], name: str = '', command: str = '', deps: str = '', outs: str = '', params: str = '', metrics: str = '', plots: str = '', wdir: str = '', force: bool = False, pipeline_id: str = '', targets: str = '', require_remote: bool = False, environment_json: str = '', dry_run: bool | None = None) -> str

Use this when you need to define or run a reproducible DVC pipeline: list stages, add/update a stage, or reproduce.

Actions — name(required, optional?): list_stages(): list DVC pipeline stages from dvc.yaml. add_stage(name, command, deps?, outs?, params?, metrics?, plots?, wdir?, force?, dry_run?): add/update a stage. run(pipeline_id, targets?, require_remote?, environment_json?, dry_run?): run/plan DVC repro and write a manifest.

dry_run defaults to true. All return JSON.

Source code in memory/artifact_tools.py

def pipeline(
    action: Literal["list_stages", "add_stage", "run"],
    name: str = "",
    command: str = "",
    deps: str = "",
    outs: str = "",
    params: str = "",
    metrics: str = "",
    plots: str = "",
    wdir: str = "",
    force: bool = False,
    pipeline_id: str = "",
    targets: str = "",
    require_remote: bool = False,
    environment_json: str = "",
    dry_run: bool | None = None,
) -> str:
    """Use this when you need to define or run a reproducible DVC pipeline: list stages, add/update a stage, or reproduce.

    Actions — name(required, optional?):
        list_stages(): list DVC pipeline stages from dvc.yaml.
        add_stage(name, command, deps?, outs?, params?, metrics?, plots?, wdir?, force?, dry_run?):
            add/update a stage.
        run(pipeline_id, targets?, require_remote?, environment_json?, dry_run?):
            run/plan DVC repro and write a manifest.

    ``dry_run`` defaults to true. All return JSON.
    """
    if action == "list_stages":
        return list_pipeline_stages()
    if action == "add_stage":
        return add_pipeline_stage(
            name=name, command=command, deps=deps, outs=outs, params=params,
            metrics=metrics, plots=plots, wdir=wdir, force=force,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "run":
        return run_pipeline(
            pipeline_id=pipeline_id, targets=targets, require_remote=require_remote,
            environment_json=environment_json,
            dry_run=(True if dry_run is None else dry_run),
        )
    return _bad_action(pipeline, action)

experiment

experiment(action: Literal['run', 'push_manifest', 'verify_manifest', 'export_catalog', 'reproduce', 'rerun', 'get_job'], experiment_id: str = '', command: str = '', artifacts_json: str = '', cwd: str = '', execute: bool = False, require_remote: bool = False, include_dirty_diff: bool = False, dirty_note: str = '', environment_json: str = '', catalog_prefix: str = '', copy_small_artifacts: bool = True, max_artifact_bytes: int = 1000000, pull: bool = False, where: str = 'cloud', executor: str = '', job_id: str = '', tail_chars: int = 8000, dry_run: bool | None = None) -> str

Use this when running or replicating an experiment: capture a provenance manifest, ship/verify it, reproduce, or rerun.

Actions — name(required, optional?): run(experiment_id, command, artifacts_json?, cwd?, execute?, require_remote?, include_dirty_diff?, dirty_note?, environment_json?, dry_run?): create a manifest and optionally execute. dry_run default true. push_manifest(experiment_id, dry_run?): push a manifest's DVC artifacts. dry_run default false. verify_manifest(experiment_id, dry_run?): verify a manifest's pointers/files. dry_run default false. export_catalog(experiment_id, catalog_prefix?, copy_small_artifacts?, max_artifact_bytes?, dry_run?): export a human-readable S3 catalog. dry_run default true. reproduce(experiment_id, pull?, execute?, dry_run?): prepare/execute reproduction. dry_run default true. rerun(experiment_id, where?, executor?, dry_run?): rerun in a clean sandbox. dry_run default true. get_job(job_id, tail_chars?): state + log tail of a cloud rerun.

All return JSON.

Source code in memory/artifact_tools.py

def experiment(
    action: Literal[
        "run", "push_manifest", "verify_manifest", "export_catalog", "reproduce", "rerun", "get_job"
    ],
    experiment_id: str = "",
    command: str = "",
    artifacts_json: str = "",
    cwd: str = "",
    execute: bool = False,
    require_remote: bool = False,
    include_dirty_diff: bool = False,
    dirty_note: str = "",
    environment_json: str = "",
    catalog_prefix: str = "",
    copy_small_artifacts: bool = True,
    max_artifact_bytes: int = 1000000,
    pull: bool = False,
    where: str = "cloud",
    executor: str = "",
    job_id: str = "",
    tail_chars: int = 8000,
    dry_run: bool | None = None,
) -> str:
    """Use this when running or replicating an experiment: capture a provenance manifest, ship/verify it, reproduce, or rerun.

    Actions — name(required, optional?):
        run(experiment_id, command, artifacts_json?, cwd?, execute?, require_remote?, include_dirty_diff?, dirty_note?, environment_json?, dry_run?):
            create a manifest and optionally execute. ``dry_run`` default true.
        push_manifest(experiment_id, dry_run?): push a manifest's DVC artifacts.
            ``dry_run`` default false.
        verify_manifest(experiment_id, dry_run?): verify a manifest's pointers/files.
            ``dry_run`` default false.
        export_catalog(experiment_id, catalog_prefix?, copy_small_artifacts?, max_artifact_bytes?, dry_run?):
            export a human-readable S3 catalog. ``dry_run`` default true.
        reproduce(experiment_id, pull?, execute?, dry_run?): prepare/execute
            reproduction. ``dry_run`` default true.
        rerun(experiment_id, where?, executor?, dry_run?): rerun in a clean sandbox.
            ``dry_run`` default true.
        get_job(job_id, tail_chars?): state + log tail of a cloud rerun.

    All return JSON.
    """
    if action == "run":
        return run_experiment(
            experiment_id=experiment_id, command=command, artifacts_json=artifacts_json,
            cwd=cwd, execute=execute, require_remote=require_remote,
            include_dirty_diff=include_dirty_diff, dirty_note=dirty_note,
            environment_json=environment_json,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "push_manifest":
        return push_manifest_artifacts(
            experiment_id=experiment_id, dry_run=(False if dry_run is None else dry_run))
    if action == "verify_manifest":
        return verify_manifest(
            experiment_id=experiment_id, dry_run=(False if dry_run is None else dry_run))
    if action == "export_catalog":
        return export_experiment_catalog(
            experiment_id=experiment_id, catalog_prefix=catalog_prefix,
            copy_small_artifacts=copy_small_artifacts, max_artifact_bytes=max_artifact_bytes,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "reproduce":
        return reproduce_experiment(
            experiment_id=experiment_id, pull=pull, execute=execute,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "rerun":
        return rerun_experiment(
            experiment_id=experiment_id, where=where, executor=executor,
            dry_run=(True if dry_run is None else dry_run),
        )
    if action == "get_job":
        return get_rerun_job(job_id=job_id, tail_chars=tail_chars)
    return _bad_action(experiment, action)

overview

overview() -> str

Use this first when you're unsure how to work with artifacts: a compact orientation map.

Read-only. Returns the artifacts server's purpose and a tool -> action index. Authoritative action values also live on each tool's own action schema enum.

Source code in memory/artifact_tools.py

def overview() -> str:
    """Use this first when you're unsure how to work with artifacts: a compact orientation map.

    Read-only. Returns the artifacts server's purpose and a tool -> action index.
    Authoritative action values also live on each tool's own ``action`` schema enum.
    """
    return (
        "# Memory-artifacts server — orientation\n\n"
        "Manages large / regenerable files via DVC (one S3 remote per repo) and\n"
        "captures experiment provenance. Requires the `[artifacts]` extra plus a\n"
        "configured remote (`artifact(action=\"configure_remote\")`).\n\n"
        "## Tools\n"
        "- `artifact(action=configure_remote|add|push|pull|verify|list|resolve|backup)` — DVC file lifecycle.\n"
        "- `pipeline(action=list_stages|add_stage|run)` — reproducible DVC pipeline stages.\n"
        "- `experiment(action=run|push_manifest|verify_manifest|export_catalog|reproduce|rerun|get_job)` — experiment manifests + reruns.\n\n"
        "Tip: prefer `dry_run=true` first for anything that touches the remote.\n"
    )

register

register(mcp_server: object, schedule_commit: Callable[[str], None] | None = None) -> None

Register the artifact tools onto mcp_server.

Parameters:

Name	Type	Description	Default
mcp_server	object	A FastMCP instance exposing .tool().	required
schedule_commit	Callable[[str], None] | None	Optional callable (filepath: str) -> None used to commit manifest/dvc.yaml writes. When omitted, commits are skipped.	None

Source code in memory/artifact_tools.py

def register(mcp_server: object, schedule_commit: Callable[[str], None] | None = None) -> None:
    """Register the artifact tools onto ``mcp_server``.

    Args:
        mcp_server: A FastMCP instance exposing ``.tool()``.
        schedule_commit: Optional callable ``(filepath: str) -> None`` used to
            commit manifest/dvc.yaml writes. When omitted, commits are skipped.
    """
    global _schedule_memory_commit
    if schedule_commit is not None:
        _schedule_memory_commit = schedule_commit
    for fn in ARTIFACT_TOOLS:
        mcp_server.tool()(fn)