Yes. In Prefect OSS 3.4.4 you can query artifacts produced by a flow via: - Python client (recommended) - REST API - CLI (quick inspection) Below are concise examples for each. Python (Prefect 3.x client) - Filter artifacts by flow run id and/or key

from prefect.client.orchestration import get_client, SyncPrefectClient
from prefect.client.schemas import filters as f

flow_run_id = "00000000-0000-0000-0000-000000000000"  # your flow run id

# Async
async def list_artifacts_for_flow_run(flow_run_id: str):
    async with get_client() as client:
        filters = f.FilterSet(
            artifacts=f.ArtifactFilter(
                flow_run_id=f.ArtifactFilterFlowRunId(any_=[flow_run_id])
                # optionally: key=f.ArtifactFilterKey(any_=["my-key"])
            )
        )
        artifacts = await client.read_artifacts(filters=filters, limit=200)
        return artifacts

# Sync
def list_artifacts_for_flow_run_sync(flow_run_id: str):
    with SyncPrefectClient() as client:
        filters = f.FilterSet(
            artifacts=f.ArtifactFilter(
                flow_run_id=f.ArtifactFilterFlowRunId(any_=[flow_run_id])
            )
        )
        artifacts = client.read_artifacts(filters=filters, limit=200)
        return artifacts

- Get the “latest per key” (artifact collections), optionally scoped to a flow run:

from prefect.client.orchestration import get_client
from prefect.client.schemas import filters as f

async def list_latest_artifacts_for_flow_run(flow_run_id: str):
    async with get_client() as client:
        col_filters = f.FilterSet(
            artifact_collections=f.ArtifactCollectionFilter(
                flow_run_id=f.ArtifactCollectionFilterFlowRunId(any_=[flow_run_id])
            )
        )
        collections = await client.read_artifact_collections(filters=col_filters, limit=200)
        return collections

Handy note inside flows: the current run id is available as

prefect.runtime.flow_run.id

. REST API - Read artifacts with filters (use POST so you can send a JSON filter body):

POST {PREFECT_API_URL}/artifacts/read
Content-Type: application/json

{
  "filters": {
    "artifacts": {
      "flow_run_id": { "any_": ["00000000-0000-0000-0000-000000000000"] },
      "key": { "any_": ["my-key"] }
    }
  },
  "limit": 200,
  "offset": 0
}

- Read a single artifact by id:

GET {PREFECT_API_URL}/artifacts/{artifact_id}

- Read artifact collections (latest per key), optionally filtered by flow_run_id:

POST {PREFECT_API_URL}/artifacts/collections/read
Content-Type: application/json

{
  "filters": {
    "artifact_collections": {
      "flow_run_id": { "any_": ["00000000-0000-0000-0000-000000000000"] }
    }
  },
  "limit": 200
}

Docs: - REST: read artifacts - REST: read artifact by ID - Python filters reference CLI - List artifacts (latest per key by default; add

--all

for all versions):

prefect artifact ls
prefect artifact ls --all --limit 200

- Inspect all versions for a given key:

prefect artifact inspect "my-key"

- Delete an artifact:

prefect artifact delete <artifact-id>

Would you prefer a Python snippet you can drop into your project, or a cURL example against your OSS server? If you share your use case (by key, by flow run id, latest-only, etc.), I can tailor the exact query.

Sorry, I encountered an error while processing your request. Please try again.

Thanks for the snippet — I think the issue is how the filters are being passed. In Prefect 3.4.4,

PrefectClient.read_artifacts

expects individual filter kwargs (e.g.,

artifact_filter=...

) rather than a single

filters=FilterSet(...)

. When

filters

isn’t recognized, the client just returns all artifacts. Try this minimal working version:

from typing import Any, Dict, List
from prefect.client.orchestration import get_client
from prefect.client.schemas import filters as f

async def get_artifacts_for_flow_run(flow_run_id: str) -> List[Dict[str, Any]]:
    async with get_client() as client:
        artifacts = await client.read_artifacts(
            limit=200,
            artifact_filter=f.ArtifactFilter(
                flow_run_id=f.ArtifactFilterFlowRunId(any_=[flow_run_id]),
                key=f.ArtifactFilterKey(any_=["my_key"]),
            ),
        )
        return artifacts

If you want only the latest artifact per key, use artifact collections instead:

async def get_latest_artifacts_for_flow_run(flow_run_id: str):
    async with get_client() as client:
        collections = await client.read_artifact_collections(
            limit=200,
            artifact_collection_filter=f.ArtifactCollectionFilter(
                flow_run_id=f.ArtifactCollectionFilterFlowRunId(any_=[flow_run_id]),
                key=f.ArtifactCollectionFilterKey(any_=["my_key"]),
            ),
        )
        return collections

REST equivalents (handy for verifying your filter works server-side): - Read artifacts

POST {PREFECT_API_URL}/artifacts/read
Content-Type: application/json

{
  "limit": 200,
  "artifact_filter": {
    "flow_run_id": { "any_": ["<your-flow-run-id>"] },
    "key": { "any_": ["my_key"] }
  }
}

- Read artifact collections (latest per key)

POST {PREFECT_API_URL}/artifacts/collections/read
Content-Type: application/json

{
  "limit": 200,
  "artifact_collection_filter": {
    "flow_run_id": { "any_": ["<your-flow-run-id>"] },
    "key": { "any_": ["my_key"] }
  }
}

References: - REST: read artifacts - REST: read artifact by ID - Python filters reference If you still see everything returned after this change, can you share: - A sample artifact’s

key

and

flow_run_id

from

prefect artifact inspect "my_key"

or from the UI - Your exact imports (they should come from

prefect.client.schemas.filters

)