You don’t need to wire DLT’s

LogCollector

into Prefect directly. In Prefect 3.x, the simplest and most reliable way to get DLT’s logs into the Prefect UI is to have Prefect capture DLT’s Python logger. Do this: 1) Tell Prefect to capture DLT’s logger - Set this in the environment where your flow runs (worker/pod/container):

export PREFECT_LOGGING_EXTRA_LOGGERS=dlt

- If you prefer to set it in code for local testing:

from prefect.settings import temporary_settings

with temporary_settings({"PREFECT_LOGGING_EXTRA_LOGGERS": ["dlt"]}):
    run_my_flow()

2) Capture prints (optional but useful) - Either set on the flow/task or globally:

@flow(log_prints=True)
def my_flow():
    ...
# or
export PREFECT_LOGGING_LOG_PRINTS=true

3) Use Prefect’s run logger for your own messages

from prefect import flow
from prefect.logging import get_run_logger
import dlt

@flow(log_prints=True)
def run_culture_counts(resource_name: str):
    logger = get_run_logger()

    cc_source = make_culture_counts_source(
        username=_load_secret("culture-counts-username"),
        password=_load_secret("culture-counts-password"),
        org_id=_load_secret("organisation-id"),
        base_url="<https://api.impactandinsight.co.uk>",
    )

    # You can drop LogCollector entirely if PREFECT_LOGGING_EXTRA_LOGGERS is set.
    pipeline = dlt.pipeline(
        pipeline_name=f"culture_counts_{resource_name}",
        destination="postgres",
        dataset_name="raw_culture_counts",
        # Optional: if LogCollector accepts a logger instance, you can forward to Prefect’s logger:
        # progress=LogCollector(logger=get_run_logger()),
    )

    <http://logger.info|logger.info>("Starting DLT pipeline run")
    load_info = pipeline.run(cc_source.with_resources(resource_name))
    load_info.raise_on_failed_jobs()
    <http://logger.info|logger.info>(f"Culture Counts {resource_name} sync complete: {load_info}")

That’s it—DLT’s logs emitted via the

dlt

logger will appear in your Prefect run logs alongside your own messages. Quick checks if you still don’t see them: - Verify the setting is in the runtime environment:

from prefect.settings import PREFECT_LOGGING_EXTRA_LOGGERS
print(PREFECT_LOGGING_EXTRA_LOGGERS.value())  # should include 'dlt'

- Ensure you’re running inside a Prefect flow/task run (logs only go to the API in a run context). - Emit a test log to confirm capture:

import logging
logging.getLogger("dlt").info("DLT test log")  # should show in the UI

Notes: - You generally only need

dlt

in

PREFECT_LOGGING_EXTRA_LOGGERS

. If DLT uses sub-loggers (e.g.,

dlt.pipeline

) they will typically propagate to

dlt

unless DLT disables propagation. - Using

LogCollector(logger="dlt_logger")

won’t help unless that logger is one Prefect is capturing. Either set

logger="dlt"

or pass

get_run_logger()

if

LogCollector

accepts a logger instance. Docs for reference: - Add logging to flows and tasks: How to add logging - Advanced logging customization (incl. EXTRA_LOGGERS): Advanced logging customization If it’s still not showing up, let me know: - How are you running the flow (worker type / environment)? - Do you see DLT logs in the process stdout but not in Prefect UI? - Can you share a short snippet of the DLT logs you expect to see?

@Marvin is there a prefect mcp server or skill I can link into claude?

Yes — there’s an official Prefect MCP server you can connect to Claude. - Repo: Prefect MCP Server - How-to guide: Use the Prefect MCP server What it does - Read-focused tools for Claude to inspect Prefect: flow/task runs, deployments, work pools, logs, health, etc. - Great for diagnosis and answering “what’s going on with my workflows?” directly from Claude. Quick connect (Claude Desktop/Code via stdio) - Add an MCP server entry pointing to the Prefect MCP server. A common setup uses uvx to run the server:

# Command the MCP client should run:
uvx --from prefect-mcp prefect-mcp-server

- Provide Prefect credentials via env vars: - Prefect Cloud: - PREFECT_API_URL=https://api.prefect.cloud/api/accounts/&lt;ACCOUNT_ID&gt;/workspaces/&lt;WORKSPACE_ID> - PREFECT_API_KEY=<YOUR_API_KEY> - Prefect OSS: - PREFECT_API_URL=http//&lt;your prefect host&gt;4200/api - (Auth optional, depending on your server setup) Example Claude config snippet (conceptual):

{
  "mcpServers": {
    "prefect": {
      "command": "uvx",
      "args": [
        "--from",
        "prefect-mcp",
        "prefect-mcp-server"
      ],
      "env": {
        "PREFECT_API_URL": "<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>>",
        "PREFECT_API_KEY": "<YOUR_API_KEY>"
      }
    }
  }
}

Write actions (trigger/cancel runs, etc.) - The MCP server is read-optimized. For mutations, let Claude invoke the Prefect CLI or REST API as a separate tool/skill: - CLI reference (verify in your shell before wiring to Claude):

prefect deploy

,

prefect deployment run ...

,

prefect flow-run cancel ...

- API docs: Prefect Cloud REST API and REST API reference - Python client: API client how-to Want me to generate a ready-to-paste Claude MCP config for your environment? If you share whether you’re on Cloud or OSS (and your account/workspace IDs for Cloud), I’ll tailor the exact snippet.

Great news — you can run exactly what we use for docs search locally. What you want - A Claude “skill” that can: - Inspect your Prefect workspace (runs, logs, deployments, etc.) - Semantically search Prefect docs with high-quality snippets - All available locally via MCP so Claude can call it on demand Use the official Prefect MCP server - Repo: prefect-mcp-server - How-to: Use the Prefect MCP server It includes: - Prefect tools: get_flow_runs, get_deployments, get_logs, dashboards, etc. - A proxied “docs” tool namespace that does semantic search over Prefect docs: - Tool name:

docs:search_prefect

- Returns ranked snippets + canonical doc links Local setup (stdio) with Claude Code/Desktop - This runs both the Prefect tools and the docs search on your machine:

claude mcp add prefect \
  -- uvx --from prefect-mcp prefect-mcp-server

- If you want it to connect to your Prefect Cloud workspace:

claude mcp add prefect \
  -e PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>> \
  -e PREFECT_API_KEY=<YOUR_API_KEY> \
  -- uvx --from prefect-mcp prefect-mcp-server

Docs search specifics - The server mounts a docs MCP under the

docs

namespace, exposing: -

docs:search_prefect(query: str, top_k: int=5)

- Under the hood: OpenAI embeddings + TurboPuffer vector DB with an ingestion pipeline that keeps the Prefect docs indexed. - You can run the docs MCP locally as-is, or fork and re-index with your own stack if you want full offline/local embeddings (see below). Run just the docs search locally (standalone) - The repo includes a standalone docs server you can run: - Path:

packages/docs_mcp_server

- Example:

git clone <https://github.com/PrefectHQ/prefect-mcp-server>
cd prefect-mcp-server/packages/docs_mcp_server
uv sync
export OPENAI_API_KEY=...
export TURBOPUFFER_API_KEY=...
uv run -m docs_mcp_server

- Then add it to Claude as an MCP server (stdio or HTTP). You’ll get the

docs:search_prefect

tool. Want fully local embeddings (no OpenAI/TurboPuffer)? - You can fork the ingestion pipeline in

packages/ingestion_pipeline

and swap: - Embeddings: sentence-transformers (e.g., all-MiniLM-L6-v2) - Vector store: FAISS on disk - Then expose a small MCP server that queries your FAISS index. The repo’s docs MCP code is a good template to modify. If you share your preferred setup (Cloud vs OSS, and whether you want online embeddings or fully local), I can give you a copy-paste Claude MCP config and/or a minimal “local-only docs search MCP” using FAISS + sentence-transformers.