Short answer: fetch the flow run’s terminal state from the API, read its result metadata to learn the S3 object key (and the storage block id), read the S3 object with boto3, then deserialize using the serializer indicated in the metadata. This avoids prefect-aws entirely. Below is a complete, robust example that: - Works in Prefect 3.4.9 - Uses only the core Prefect client + boto3 - Handles both “combined” result files (default) and “separate” result/metadata layouts

import json
import uuid
import boto3
from typing import Any
from prefect.client.orchestration import get_client
from prefect.exceptions import UnfinishedRun, MissingResult

# Optional: only used to parse "combined" result files in the default layout
from prefect._result_records import ResultRecord

def read_flow_result_from_s3_without_prefect_aws(flow_run_id: str) -> Any:
    """
    Given a Prefect 3 flow run ID, download its persisted result from S3
    and return the deserialized Python object. Requires no prefect-aws;
    uses boto3 directly.

    Assumptions:
      - The flow’s result persistence used an S3 storage block
      - Your environment (e.g., CodeBuild role) can GetObject from that bucket
    """
    fid = uuid.UUID(flow_run_id)

    # 1) Read the flow run and its terminal state
    with get_client(sync_client=True) as client:
        flow_run = client.read_flow_run(fid)

        if not flow_run.state:
            raise MissingResult(f"No state found for flow run {flow_run_id}")

        if not flow_run.state.is_final():
            raise UnfinishedRun(f"Flow run {flow_run_id} is not in a final state: {flow_run.state.type}")

        # This is a ResultRecordMetadata when retrieved from the API
        metadata = flow_run.state.data
        if not metadata or not getattr(metadata, "storage_key", None):
            raise MissingResult("This run does not have a persisted result (no storage key found).")

        storage_key = metadata.storage_key
        storage_block_id = getattr(metadata, "storage_block_id", None)
        if storage_block_id is None:
            raise MissingResult(
                "Result metadata does not reference a storage block; cannot determine S3 bucket."
            )

        # 2) Look up the storage block document to find the S3 bucket info
        #    (No prefect-aws needed; we just read the raw block document JSON.)
        block_doc = client.read_block_document(storage_block_id, include_secrets=False)
        bucket_name = block_doc.data.get("bucket_name")
        # Note: storage_key already includes any bucket_folder/prefix used by the block.
        if not bucket_name:
            raise RuntimeError("S3 bucket_name not found in storage block data.")

    # 3) Download the object from S3 (boto3 will use CodeBuild role by default)
    s3 = boto3.client("s3")
    resp = s3.get_object(Bucket=bucket_name, Key=storage_key)
    body = resp["Body"].read()

    # 4) Deserialize
    # Prefect supports two layouts:
    #  - Combined (default): one JSON file containing both metadata and the serialized result
    #  - Separate: the object at `storage_key` is raw serialized bytes (pickle/json), and
    #    metadata is stored separately. In that case we use the serializer from state metadata.

    # Try combined layout first
    try:
        # This succeeds only if 'body' is JSON of a ResultRecord
        # Using Prefect’s helper keeps serializer handling consistent
        result = ResultRecord.deserialize(body).result
        return result
    except Exception:
        # Fallback: treat 'body' as raw serialized result bytes
        # and use the serializer declared in the state metadata.
        # metadata.serializer has .loads() that accepts bytes
        return metadata.serializer.loads(body)

Notes and gotchas: - You need AWS IAM permission to s3:GetObject (and to decrypt if KMS-encrypted). - If the run is not in a final state or was not configured to persist results, you’ll get UnfinishedRun or MissingResult.

- The storage block’s raw fields come from the saved S3 block document; for S3 they’re typically: - bucket_name: the bucket - bucket_folder: a default prefix (already included in metadata.storage_key) - You do not need prefect-aws for this—reading the block document data and the serializer is handled by core Prefect; the S3 read is done with boto3. If you prefer a one-liner with Prefect doing the download/deserialization for you,

flow_run.state.result()

works—however it requires the S3 block implementation from prefect-aws to be installed. References: - Results (concepts): Prefect 3 docs: Results - Blocks (concepts): Prefect 3 docs: Blocks If you share a specific flow_run_id example, I can help you verify whether the storage is “combined” or “separate” and adjust the code accordingly.

EOFError: Ran out of input almost always means pickle tried to load b'' (zero bytes). Since run.state.result() works, the most likely culprit is the object key you’re reading with boto3 is not the same key Prefect’s storage block reads. Most common cause - S3Bucket blocks prepend a bucket-level prefix (bucket_folder) when reading/writing. - metadata.storage_key is the “path” Prefect passes to the block, not necessarily the final S3 object key. - run.state.result() works because the block adds bucket_folder under the hood; your direct boto3 call likely skipped that. Try this adjusted version that: - Prepends bucket_folder when present - Verifies ContentLength via head_object - Handles both “combined” and “separate” result layouts - Reads the streaming body exactly once

import json
import uuid
import boto3
from typing import Any
from prefect.client.orchestration import get_client
from prefect.exceptions import UnfinishedRun, MissingResult
from prefect._result_records import ResultRecord

def _join_s3_key(*parts: str) -> str:
    clean = [p.strip("/") for p in parts if p and p.strip("/")]
    return "/".join(clean)

def read_flow_result_from_s3_without_prefect_aws(flow_run_id: str) -> Any:
    fid = uuid.UUID(flow_run_id)

    with get_client(sync_client=True) as client:
        flow_run = client.read_flow_run(fid)

        if not flow_run.state:
            raise MissingResult(f"No state found for flow run {flow_run_id}")
        if not flow_run.state.is_final():
            raise UnfinishedRun(f"Flow run {flow_run_id} is not final: {flow_run.state.type}")

        metadata = flow_run.state.data
        if not metadata or not getattr(metadata, "storage_key", None):
            raise MissingResult("This run does not have a persisted result (no storage key found).")

        storage_key = metadata.storage_key
        storage_block_id = getattr(metadata, "storage_block_id", None)
        if storage_block_id is None:
            raise MissingResult("Result metadata does not reference a storage block; cannot determine S3 bucket.")

        # Read the storage block document to get bucket name + folder/prefix
        block_doc = client.read_block_document(storage_block_id, include_secrets=False)
        data = block_doc.data or {}
        bucket_name = data.get("bucket_name") or data.get("bucket")  # be tolerant to field name variations
        bucket_folder = data.get("bucket_folder", "") or data.get("key_prefix", "") or data.get("bucket_path", "")

        if not bucket_name:
            raise RuntimeError("S3 bucket name not found in storage block document data.")

    # Construct the final S3 object key as the block would
    s3_key = _join_s3_key(bucket_folder, storage_key)

    s3 = boto3.client("s3")

    # Sanity check: object size
    head = s3.head_object(Bucket=bucket_name, Key=s3_key)
    size = head.get("ContentLength", 0)
    if size == 0:
        raise MissingResult(f"S3 object is empty: s3://{bucket_name}/{s3_key}")

    # Download (read once!)
    resp = s3.get_object(Bucket=bucket_name, Key=s3_key)
    body = resp["Body"].read()

    # Try combined layout first (single JSON file with both metadata and result)
    try:
        return ResultRecord.deserialize(body).result
    except Exception:
        # Fallback: 'separate' layout — body is raw serialized result; use serializer from state metadata
        return metadata.serializer.loads(body)

If this still returns empty: - Double-check you’re not reading the streaming body twice. The stream is consumable; a second read() returns b''. - Print/log the exact bucket/key and compare to what Prefect is reading: - bucket: block_doc.data.get("bucket_name") or "bucket" - folder/prefix: block_doc.data.get("bucket_folder") or "key_prefix"/"bucket_path" - storage_key: metadata.storage_key - final key: f"{bucket_folder}/{storage_key}" with single slashes

- Confirm head_object returns a non-zero ContentLength. If it’s zero but run.state.result() works, you’re likely pointing at the wrong key (e.g., missing prefix). Optional quick test - Try JSON-loading the object to see if it’s the combined file: - If json.loads(body) has keys "metadata" and "result", you’ve got the combined layout and the content is correct. - If json.loads fails and size > 0, it’s likely the raw serialized result; metadata.serializer.loads(body) should succeed. If you can share a redacted example of: - metadata.storage_key - block_doc.data (just the bucket-related fields) - the s3:// URL you attempted I can help pinpoint exactly where the path composition is going wrong.

That byte string is base64, not a raw pickle. If you pass it directly to pickle (or to Prefect’s PickleSerializer.loads), you’ll get EOFError because the unpickler is expecting binary starting with 0x80 0x05, not the ASCII “gAWV…”. Quick sanity check in your REPL:

import base64, pickle
print(pickle.loads(base64.b64decode(b"gAWVDwAAAAAAAACMC2Zsb3ctb3V0cHV0lC4=")))
# -> should print your original string (e.g., "flow-output")

Why you’re seeing this: - Prefect’s “combined” result file is JSON where the result bytes are base64-encoded. - Even in the “separate” layout, some paths still yield a base64-encoded representation when the result is later embedded in JSON. - The canonical path (run.state.result()) knows which serializer was used and decodes appropriately. Reading the S3 object directly requires one extra step: base64-decode if the payload is base64 text, then deserialize with the serializer in the state metadata. Drop-in fix for your S3 reader:

import base64
import uuid
import boto3
from prefect.client.orchestration import get_client
from prefect._result_records import ResultRecord
from prefect.exceptions import UnfinishedRun, MissingResult

def read_flow_result_from_s3(flow_run_id: str):
    fid = uuid.UUID(flow_run_id)

    with get_client(sync_client=True) as client:
        fr = client.read_flow_run(fid)
        if not fr.state:
            raise MissingResult("No state on flow run")
        if not fr.state.is_final():
            raise UnfinishedRun(f"Run not final: {fr.state.type}")

        metadata = fr.state.data  # ResultRecordMetadata from API
        if not getattr(metadata, "storage_key", None):
            raise MissingResult("No storage_key on state")

        block_doc = client.read_block_document(metadata.storage_block_id, include_secrets=False)
        bucket = block_doc.data.get("bucket_name")
        prefix = (block_doc.data.get("bucket_folder") or "").strip("/")
        key = "/".join([p for p in [prefix, metadata.storage_key] if p])

    s3 = boto3.client("s3")
    body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()

    # 1) Try combined JSON (ResultRecord) first
    try:
        return ResultRecord.deserialize(body).result
    except Exception:
        pass

    # 2) If the payload is base64 text, decode it, then use the state’s serializer
    try:
        decoded = base64.b64decode(body, validate=True)
        if decoded:  # non-empty after decode
            return metadata.serializer.loads(decoded)
    except Exception:
        pass

    # 3) Otherwise, treat the body as already the raw serialized bytes
    return metadata.serializer.loads(body)

Tips: - Your sample b'gAWVDwAAAAAAAACMC2Zsb3ctb3V0cHV0lC4=\n' will work with the base64 + serializer path above. - Newlines are fine; base64.b64decode handles them. - If you ever hit JSON results (combined layout), ResultRecord.deserialize(body).result will succeed immediately. If you want, paste the first ~40 bytes of the raw S3 payload (or confirm whether it starts with “{” vs “gA…=”), and I can tell you exactly which branch is in play.