Based on the research, here's how you can get metadata for flow runs using the Prefect client in 3.x: 1. For a specific flow run (when you know the ID):

python
from prefect.client import get_client

async def get_flow_run_metadata():
    client = get_client()
    # Replace with your flow run ID
    flow_run = await client.read_flow_run("your-flow-run-id")
    
    # Access various metadata attributes
    print(f"Flow Run Name: {flow_run.name}")
    print(f"State: {flow_run.state.type}")
    print(f"Start Time: {flow_run.start_time}")
    print(f"Tags: {flow_run.tags}")

2. For multiple flow runs with filtering:

python
from prefect.client import get_client
from prefect.client.schemas.filters import FlowRunFilter, FlowRunFilterTags

async def get_multiple_flow_runs():
    client = get_client()
    
    # Example: Get flow runs with specific tags
    flows = await client.read_flow_runs(
        flow_run_filter=FlowRunFilter(
            tags=FlowRunFilterTags(any_=["your-tag"])
        )
    )
    
    for flow in flows:
        print(f"Flow Run ID: {flow.id}")
        print(f"Name: {flow.name}")
        print(f"State: {flow.state.type}")

The

FlowRun

object contains various metadata fields including: - `id`: The flow run's unique identifier - `name`: The name of the flow run - `tags`: Any tags associated with the flow run - `state`: Current state information - `start_time`: When the flow run started - `expected_start_time`: Scheduled start time - `total_run_time`: Total runtime duration - `deployment_id`: Associated deployment ID (if any) - `parameters`: Flow parameters used in the run You can access any of these attributes from the returned

FlowRun

object. For more complex queries, you can use various filters available in

prefect.client.schemas.filters

to narrow down the flow runs based on: - State type - Deployment ID - Work queue name - Start time - Tags - And more The links to the official documentation provide more details about all available filtering options and the complete API reference.

Based on the error patterns I'm seeing, there are a few potential issues that could be causing your Pydantic error. Let me help you fix this. Here's the corrected version of the code:

python
from prefect.client import get_client
from uuid import UUID  # Add this import

async def get_flow_run_metadata(flow_run_id: str):
    client = get_client()
    # Convert string ID to UUID if needed
    if isinstance(flow_run_id, str):
        flow_run_id = UUID(flow_run_id)
        
    flow_run = await client.read_flow_run(flow_run_id)
    
    # Access various metadata attributes
    metadata = {
        "name": flow_run.name,
        "state": flow_run.state.type if flow_run.state else None,
        "start_time": flow_run.start_time,
        "tags": flow_run.tags
    }
    return metadata

# If you're calling this from a synchronous context, use:
from prefect.utilities.asyncutils import run_sync

def get_flow_run_metadata_sync(flow_run_id: str):
    return run_sync(get_flow_run_metadata)(flow_run_id)

Key changes and notes: 1. The flow run ID needs to be a valid UUID. The code now handles string conversion. 2. Added proper error handling for the metadata attributes. 3. If you're getting a specific Pydantic error, please share the exact error message. Common issues include: - Invalid flow run ID format - Missing or null values in the response - Version mismatches between client and server 4. If you're running this inside a flow, you can get the current flow run ID like this:

python
from prefect import flow
from prefect.context import get_run_context

@flow
def my_flow():
    context = get_run_context()
    flow_run_id = context.flow_run.id
    metadata = get_flow_run_metadata_sync(flow_run_id)
    print(metadata)

If you're still getting a Pydantic error, please share: 1. The exact error message 2. Your Prefect version (

prefect version

) 3. Whether you're running this code inside a flow or standalone This will help me provide a more specific solution for your case.