Connect AI agents to the Prefect MCP server

Vendor MCP

The Prefect MCP connector routes your AI agent's tool calls to Prefect's own MCP server through Scalekit. Each user signs in to Prefect once, and Scalekit stores and refreshes their tokens, so your agent never handles credentials. It comes with 15 tools.

Tools: 15
What they doRead · write · destructive: 15 · 0 · 015 read0 write0 destructive
Users sign in with: OAuth
OAuth app: Your own Prefect MCP server app

Setup

Install the SDK
- Node.js
- Python
Terminal window
npm install @scalekit-sdk/node dotenv
Terminal window
pip install scalekit-sdk-python python-dotenv
Set your credentials
Add your Scalekit credentials to your .env file. Find values in app.scalekit.com > Developers > API Credentials.
.env
```
SCALEKIT_ENVIRONMENT_URL=<your-environment-url>
SCALEKIT_CLIENT_ID=<your-client-id>
SCALEKIT_CLIENT_SECRET=<your-client-secret>
```
Create the Prefect MCP connection
In AgentKit > Connections, create a Prefect MCP connection and copy its redirect URI. The name you give it is the connection_name your code passes. See Configure connections.
Register an OAuth app
Prefect MCP server connections use your own OAuth app. Register one with Prefect MCP server and add the redirect URI you copied.
Then enter the app's Client ID and Client Secret on the Prefect MCP connection.

Authorize a user and make your first call

Node.js
Python

1
import { ScalekitClient } from '@scalekit-sdk/node'
2
import 'dotenv/config'
3
import { createInterface } from 'node:readline/promises'
4

5
const scalekit = new ScalekitClient(
6
  process.env.SCALEKIT_ENVIRONMENT_URL,
7
  process.env.SCALEKIT_CLIENT_ID,
8
  process.env.SCALEKIT_CLIENT_SECRET,
9
)
10
const actions = scalekit.actions
11

12
const connector = 'prefectmcp'
13
const identifier = 'user_123'
14

15
// Generate an authorization link for the user
16
const { link } = await actions.getAuthorizationLink({ connectionName: connector, identifier })
17
console.log('Authorize Prefect MCP:', link)
18
const rl = createInterface({ input: process.stdin, output: process.stdout })
19
await rl.question('Press Enter after authorizing...')
20
rl.close()
21

22
// Make your first call
23
const result = await actions.executeTool({
24
  connector,
25
  identifier,
26
  toolName: 'prefectmcp_docs_get_release_notes',
27
  toolInput: {},
28
})
29
console.log(result)

npx tsx quickstart.mts

1
import os
2
from scalekit import ScalekitClient
3
from dotenv import load_dotenv
4
load_dotenv()
5

6
scalekit_client = ScalekitClient(
7
    env_url=os.getenv("SCALEKIT_ENVIRONMENT_URL"),
8
    client_id=os.getenv("SCALEKIT_CLIENT_ID"),
9
    client_secret=os.getenv("SCALEKIT_CLIENT_SECRET"),
10
)
11
actions = scalekit_client.actions
12

13
connection_name = "prefectmcp"
14
identifier = "user_123"
15

16
# Generate an authorization link for the user
17
link_response = actions.get_authorization_link(
18
    connection_name=connection_name,
19
    identifier=identifier,
20
)
21
print("Authorize Prefect MCP:", link_response.link)
22
input("Press Enter after authorizing...")
23

24
# Make your first call
25
result = actions.execute_tool(
26
    tool_input={},
27
    tool_name="prefectmcp_docs_get_release_notes",
28
    connection_name=connection_name,
29
    identifier=identifier,
30
)
31
print(result)

python quickstart.py

Each user signs in once. See Authorize a user for the full flow and statuses.

Tools

Pass the exact name to execute_tool

Try in Playground Request a tool

prefectmcp_docs_get_release_notesGet authoritative, structured release notes for a Prefect OSS release.Read-only
Get Prefect Release Notes
Get authoritative, structured release notes for a Prefect OSS release. Accepts "latest" or an exact version and returns the release title, date, Markdown notes, and source URL.
Inputs
versionstring
'latest' for the latest stable Prefect OSS release, or an exact version such as '3.7.8'default latest
prefectmcp_docs_search_prefectSearch Prefect documentation for concepts, examples, and best practices.Read-only
Search Prefect Documentation
Search Prefect documentation for concepts, examples, and best practices. Returns ranked excerpts from Prefect's knowledge base for a natural-language query.
Inputs
querystringrequired
a search query to find relevant document excerpts from Prefect's knowledgebase
top_kinteger
How many document excerpts to return.
prefectmcp_get_automationsGet automations with optional filters.Read-only
Get Automations
Get automations with optional filters. Returns compact summaries by default (trigger_type, action_count). Filter by specific ID(s) for full detail including trigger config, actions, actions_on_trigger, and actions_on_resolve. Filter operators: - id.any_: Match specific automation IDs - name.any_: Match automation names - enabled.eq_: Filter by enabled state Examples: - List all automations: get_automations() - Full detail: get_automations(filter={"id": {"any_": ["<automation-id>"]}}) - Get by name: get_automations(filter={"name": {"any_": ["my-automation"]}}) - Only enabled: get_automations(filter={"enabled": {"eq_": True}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect AutomationFilter fields.
limitinteger
Maximum number of automations to return.default 100
prefectmcp_get_dashboardGet a high-level dashboard overview of the Prefect instance.Read-only
Get Dashboard
Get a high-level dashboard overview of the Prefect instance. Returns current flow run statistics, work pool status, and all active concurrency limits (global/tag-based, deployment, work pool, and work queue). Essential for diagnosing flow run delays and bottlenecks.
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
prefectmcp_get_deploymentsGet deployments with optional filters.Read-only
Get Deployments
Get deployments with optional filters. Returns compact summaries by default. Filter by specific ID(s) for full detail including parameters, parameter_openapi_schema, job_variables, work_pool details, and recent_runs. The response includes truncated=true when more matching records exist. Filter operators: - any_: Match any value in list - all_: Match all values - like_: SQL LIKE pattern matching - not_any_: Exclude values - is_null_: Check for null/not null - eq_/ne_: Equality comparisons Examples: - List all deployments: get_deployments() - Full detail: get_deployments(filter={"id": {"any_": ["<deployment-id>"]}}) - Active deployments: get_deployments(filter={"paused": {"eq_": False}}) - Production deployments: get_deployments(filter={"tags": {"all_": ["production"]}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect DeploymentFilter fields.
limitinteger
Maximum number of deployments to return.default 50
prefectmcp_get_flow_run_logsGet execution logs for a flow run.Read-only
Get Flow Run Logs
Get execution logs for a flow run. Retrieves log entries from the flow run execution, including timestamps, log levels, and messages. Examples: - Get logs: get_flow_run_logs(flow_run_id="...") - Get more logs: get_flow_run_logs(flow_run_id="...", limit=500)
Inputs
flow_run_idstringrequired
UUID of the flow run to get logs for
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
limitinteger
Maximum number of log entries to return.default 100
prefectmcp_get_flow_runsGet flow runs with optional filters.Read-only
Get Flow Runs
Get flow runs with optional filters. Returns compact summaries by default. Filter by specific ID(s) for full detail including parameters, inlined deployment info, and work pool info. The response includes truncated=true when more matching records exist. Filter operators: - any_: Match any value in list - all_: Match all values - like_: SQL LIKE pattern matching - not_any_: Exclude values - is_null_: Check for null/not null - after_/before_: Time comparisons - gt_/gte_/lt_/lte_: Numeric comparisons Examples: - List recent runs: get_flow_runs() - Get specific run: get_flow_runs(filter={"id": {"any_": ["<flow-run-id>"]}}) - Failed runs: get_flow_runs(filter={"state": {"type": {"any_": ["FAILED"]}}}) - Production runs: get_flow_runs(filter={"tags": {"all_": ["production"]}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect FlowRunFilter fields.
limitinteger
Maximum number of flow runs to return.default 50
prefectmcp_get_flowsGet flows with optional filters.Read-only
Get Flows
Get flows with optional filters. Returns a list of flows registered in the workspace. The response includes truncated=true when more matching records exist. Filter operators: - any_: Match any value in list - like_: SQL LIKE pattern matching - all_: Match all values Examples: - List all flows: get_flows() - Get specific flow: get_flows(filter={"id": {"any_": ["<flow-id>"]}}) - Flows by name pattern: get_flows(filter={"name": {"like_": "etl-%"}}) - Flows by tags: get_flows(filter={"tags": {"all_": ["production"]}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect FlowFilter fields.
limitinteger
Maximum number of flows to return.default 50
prefectmcp_get_identityGet identity and connection information for the current Prefect instance.Read-only
Get Identity
Get identity and connection information for the current Prefect instance. Returns API URL, type (cloud/oss), and user information if available. Essential for understanding which Prefect instance you're connected to.
Inputs
workspace_idstring
Prefect Cloud workspace ID. Required when using Prefect Cloud OAuth mode.
prefectmcp_get_object_schemaGet a schema for an object type.Read-only
Get Object Schema
Get a schema for an object type. An action_type narrows all three action lists and includes only referenced definitions. Trigger variants and automation guidance remain available.
Inputs
object_typestringrequired
Name of the object type to get a schema forone of automation
action_typestring
Return a schema for only this automation action type; omit for all action types
prefectmcp_get_task_runsGet task runs with optional filters.Read-only
Get Task Runs
Get task runs with optional filters. Returns a list of task runs and their details matching the filters. Note that 'task_inputs' contains dependency tracking information (upstream task relationships), not the actual parameter values passed to the task. The response includes truncated=true when more matching records exist. Filter operators: - any_: Match any value in list - like_: SQL LIKE pattern matching - not_any_: Exclude values - is_null_: Check for null/not null Examples: - List recent tasks: get_task_runs() - Get specific task: get_task_runs(filter={"id": {"any_": ["<task-run-id>"]}}) - Failed tasks: get_task_runs(filter={"state": {"type": {"any_": ["FAILED"]}}}) - Tasks by pattern: get_task_runs(filter={"name": {"like_": "%process%"}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect TaskRunFilter fields.
limitinteger
Maximum number of task runs to return.default 50
prefectmcp_get_work_poolsGet work pools with optional filters.Read-only
Get Work Pools
Get work pools with optional filters. Returns compact summaries by default (name, type, status, concurrency_limit). Filter by specific ID(s) for full detail including work queues, active worker counts, and descriptions. Essential for debugging deployment issues related to flow runs being stuck or not starting. Filter operators: - any_: Match any value in list - like_: SQL LIKE pattern matching Examples: - List all pools: get_work_pools() - Full detail: get_work_pools(filter={"id": {"any_": ["<work-pool-id>"]}}) - Kubernetes pools: get_work_pools(filter={"type": {"any_": ["kubernetes"]}})
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
filterobject
JSON filter object for advanced querying. Supports all Prefect WorkPoolFilter fields.
limitinteger
Maximum number of work pools to return.default 50
prefectmcp_list_authorized_workspacesList Prefect Cloud workspaces selected during OAuth consent.Read-only
List Authorized Workspaces
List Prefect Cloud workspaces selected during OAuth consent. Returns account handles, workspace handles, workspace IDs, and grant metadata for the workspaces available to this connection. Only available in Prefect Cloud OAuth mode.
Inputs
This tool takes no inputs.
prefectmcp_orientationReturn an overview of Prefect MCP capabilities and access boundaries.Read-only
Orientation
Return an overview of Prefect MCP capabilities and access boundaries. Summarizes supported inspection, documentation, schema, and optional authoring capabilities. It does not access or modify Prefect data.
Inputs
This tool takes no inputs.
prefectmcp_read_eventsRead and filter events from the Prefect instance.Read-only
Read Events
Read and filter events from the Prefect instance. Provides a structured view of events with filtering capabilities. Note: When no time range is specified, events from the last 1 hour are returned by default. Use occurred_after/occurred_before parameters to query a different time range. Common event type prefixes: - prefect.flow-run: Flow run lifecycle events - prefect.deployment: Deployment-related events - prefect.work-queue: Work queue events - prefect.agent: Agent events Examples: - Recent flow run events: read_events(event_type_prefix="prefect.flow-run") - Last 24 hours: read_events(occurred_after="<ISO8601-timestamp-24-hours-ago>") - Specific time range: read_events(occurred_after="2024-01-01T00:00:00Z", occurred_before="2024-01-02T00:00:00Z")
Inputs
workspace_idstringrequired
Prefect Cloud workspace ID. Required on every call to this tool for this connection (Prefect Cloud OAuth mode).
event_type_prefixstring
Filter events by type prefix
limitinteger
Maximum number of events to return.default 50
occurred_afterstring
ISO 8601 timestamp to filter events after
occurred_beforestring
ISO 8601 timestamp to filter events before

Setup

Install the SDK

Set your credentials

Create the Prefect MCP connection

Register an OAuth app

Authorize a user and make your first call

Tools

Call it

Returns

Get Prefect Release Notes

Inputs

Search Prefect Documentation

Inputs

Get Automations

Inputs

Get Dashboard

Inputs

Get Deployments

Inputs

Get Flow Run Logs

Inputs

Get Flow Runs

Inputs

Get Flows

Inputs

Get Identity

Inputs

Get Object Schema

Inputs

Get Task Runs

Inputs

Get Work Pools

Inputs

List Authorized Workspaces

Inputs

Orientation

Inputs

Read Events

Inputs

Related: other Automation connectors