airbyte_ops_mcp.mcp.prod_db_ops
MCP tools for querying the Airbyte Cloud Prod DB Replica.
This module provides MCP tools that wrap the query functions from airbyte_ops_mcp.prod_db_access.queries for use by AI agents.
MCP reference
MCP primitives registered by the prod_db_ops module of the airbyte-internal-ops server: 20 tool(s), 0 prompt(s), 0 resource(s).
Tools (20)
query_connector_pin_stats
Hints: read-only · idempotent · open-world
Query connector versions that have at least one scoped configuration pin.
Returns versions from the prod DB that are referenced by at least one
scoped_configuration pin (key = 'connector_version'). Each version
appears exactly once with per-scope pin breakdown (actor, workspace, org).
If neither filter is provided, returns the global superset across all connectors.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_definition_id |
string | null |
no | null |
Connector definition UUID to filter by (optional). Mutually exclusive with connector_canonical_name. |
connector_canonical_name |
string | null |
no | null |
Connector canonical name (e.g. source-postgres) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with connector_definition_id. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector definition UUID to filter by (optional). Mutually exclusive with `connector_canonical_name`."
},
"connector_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector canonical name (e.g. `source-postgres`) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with `connector_definition_id`."
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"description": "A connector version that has at least one scoped configuration pin.",
"properties": {
"version_id": {
"description": "The actor_definition_version UUID",
"type": "string"
},
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_name": {
"description": "Human-readable connector name",
"type": "string"
},
"docker_repository": {
"description": "Docker repository path",
"type": "string"
},
"docker_image_tag": {
"description": "Docker image tag for this version",
"type": "string"
},
"last_published": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "ISO timestamp when this version was last published"
},
"pin_count": {
"description": "Total number of scoped_configuration rows pinning to this version",
"type": "integer"
},
"breaking_change_pins": {
"default": 0,
"description": "Number of actor-scoped pins created by breaking changes",
"type": "integer"
},
"rollout_pins": {
"default": 0,
"description": "Number of pins created by connector rollouts",
"type": "integer"
},
"actor_pins": {
"description": "Number of actor-scoped pins (excludes breaking change and rollout pins)",
"type": "integer"
},
"workspace_pins": {
"description": "Number of workspace-scoped pins",
"type": "integer"
},
"org_pins": {
"description": "Number of organization-scoped pins",
"type": "integer"
}
},
"required": [
"version_id",
"connector_definition_id",
"connector_name",
"docker_repository",
"docker_image_tag",
"pin_count",
"actor_pins",
"workspace_pins",
"org_pins"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_connector_population_summary
Hints: read-only · idempotent · open-world
Summarize the applied vs potential pinning audience for a connector, by tier.
Answers "how many actors are pinned and how many are eligible for pinning, split by tier" — the population view analogous to a rollout's audience.
active: the potential audience — non-tombstoned actors of the definition that have at least one active connection (connection.status = 'active'). Inactive/disabled and deprecated connections are excluded, so this reflects the enabled, rollout-touchable population rather than every actor ever created.pinned_any: active actors that already have an effectiveconnector_versionpin at any scope (actor/workspace/org).eligible:activeminuspinned_any— actors available to pin.pinned_to_version: the applied audience for the requested version (only when a version identifier was provided).
Backed by scoped_configuration + actor/connection tables, so it is cheap
to compute. Accepts a version identifier (preferred, adds
pinned_to_version) or a definition-level identifier.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_version_id |
string | null |
no | null |
Connector version UUID. When provided, the applied audience (pinned_to_version) is included. Provide this OR connector_name + connector_version OR a definition-level identifier. |
connector_name |
string | null |
no | null |
Canonical connector name (e.g. source-postgres). Used with connector_version to resolve the version UUID. |
connector_version |
string | null |
no | null |
Semver version tag (e.g. 0.3.59). Used with connector_name. |
connector_definition_id |
string | null |
no | null |
Connector definition UUID for a definition-level summary (no pinned_to_version breakdown). |
connector_canonical_name |
string | null |
no | null |
Canonical connector name resolved to a definition ID via the registry, for a definition-level summary. |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Which customer tiers to count. Defaults to TIER_2; pass ALL to include TIER_0/TIER_1 (revenue-critical) customers. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector version UUID. When provided, the applied audience (`pinned_to_version`) is included. Provide this OR `connector_name` + `connector_version` OR a definition-level identifier."
},
"connector_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical connector name (e.g. `source-postgres`). Used with `connector_version` to resolve the version UUID."
},
"connector_version": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Semver version tag (e.g. `0.3.59`). Used with `connector_name`."
},
"connector_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector definition UUID for a definition-level summary (no `pinned_to_version` breakdown)."
},
"connector_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical connector name resolved to a definition ID via the registry, for a definition-level summary."
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Which customer tiers to count. Defaults to `TIER_2`; pass `ALL` to include TIER_0/TIER_1 (revenue-critical) customers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
}
},
"type": "object"
}
Show output JSON schema
{
"description": "Applied vs potential pinning audience for a connector, split by tier.",
"properties": {
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The version UUID, when a specific version was requested"
},
"docker_repository": {
"description": "Docker repository path",
"type": "string"
},
"docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker image tag, when a version was requested"
},
"customer_tier_filter": {
"description": "Tier filter applied to the counts (`TIER_0`/`TIER_1`/`TIER_2`/`ALL`)",
"type": "string"
},
"active": {
"description": "Potential audience: enabled actors of the definition (those with at least one active connection, `status = 'active'`), by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"pinned_any": {
"description": "Active actors already pinned to any version, by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"eligible": {
"description": "Active actors not pinned to any version (available to pin), by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"pinned_to_version": {
"anyOf": [
{
"description": "Summary of tier distribution across a set of results.",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
{
"type": "null"
}
],
"default": null,
"description": "Applied audience: actors pinned to the requested version, by tier. `None` when no specific version was requested."
}
},
"required": [
"connector_definition_id",
"docker_repository",
"customer_tier_filter",
"active",
"pinned_any",
"eligible"
],
"type": "object"
}
query_connector_version_health_summary
Hints: read-only · idempotent · open-world
Summarize actor health for a connector version into four buckets.
Answers "how many actors on a version are healthy / unhealthy / awaiting / disabled". Classification per actor over the lookback window:
healthy: at least one successful sync (the same success signal the autopilot health gate uses).unhealthy: failures and no successes.awaiting: ran but produced only non-terminal jobs (no result yet).disabled: (wheninclude_pinned_disabled) pinned to the version with no jobs at all in the window — the dormant/inactive audience.
Built on the attempt/version primitive — the version stamped into
jobs.config at job-creation time — not the current pin state, so it
reflects actors that actually ran this version. This scans jobs over the
window and is more expensive than the population summary; keep days
bounded and query per-version.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_version_id |
string | null |
no | null |
Connector version UUID. Provide this OR connector_name + connector_version. |
connector_name |
string | null |
no | null |
Canonical connector name (e.g. source-postgres). Used with connector_version to resolve the version UUID. |
connector_version |
string | null |
no | null |
Semver version tag (e.g. 0.3.59). Used with connector_name. |
days |
integer |
no | 7 |
Number of days to look back (default: 7, max: 30) |
include_pinned_disabled |
boolean |
no | true |
If True (default), actors pinned to the version that ran no jobs in the window are counted as disabled. |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Which customer tiers to count. Defaults to TIER_2; pass ALL to include TIER_0/TIER_1 (revenue-critical) customers. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector version UUID. Provide this OR `connector_name` + `connector_version`."
},
"connector_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical connector name (e.g. `source-postgres`). Used with `connector_version` to resolve the version UUID."
},
"connector_version": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Semver version tag (e.g. `0.3.59`). Used with `connector_name`."
},
"days": {
"default": 7,
"description": "Number of days to look back (default: 7, max: 30)",
"maximum": 30,
"minimum": 1,
"type": "integer"
},
"include_pinned_disabled": {
"default": true,
"description": "If `True` (default), actors pinned to the version that ran no jobs in the window are counted as `disabled`.",
"type": "boolean"
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Which customer tiers to count. Defaults to `TIER_2`; pass `ALL` to include TIER_0/TIER_1 (revenue-critical) customers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
}
},
"type": "object"
}
Show output JSON schema
{
"description": "Four-bucket health rollup for the actors running a connector version.",
"properties": {
"connector_version_id": {
"description": "The connector version UUID",
"type": "string"
},
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"docker_repository": {
"description": "Docker repository path",
"type": "string"
},
"docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker image tag for this version"
},
"days": {
"description": "Lookback window in days",
"type": "integer"
},
"customer_tier_filter": {
"description": "Tier filter applied to the counts (`TIER_0`/`TIER_1`/`TIER_2`/`ALL`)",
"type": "string"
},
"healthy": {
"description": "Actors with at least one successful sync",
"type": "integer"
},
"unhealthy": {
"description": "Actors with failures and no successes in the window",
"type": "integer"
},
"awaiting": {
"description": "Actors that ran but produced only non-terminal jobs (no result yet)",
"type": "integer"
},
"disabled": {
"description": "Actors pinned to the version that produced no jobs in the window \u2014 the dormant/inactive audience",
"type": "integer"
},
"total_actors": {
"description": "Total actors counted across all states",
"type": "integer"
},
"healthy_by_tier": {
"description": "Healthy actors by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"unhealthy_by_tier": {
"description": "Unhealthy actors by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"awaiting_by_tier": {
"description": "Awaiting-results actors by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
},
"disabled_by_tier": {
"description": "Disabled actors by tier",
"properties": {
"tier_0_count": {
"default": 0,
"description": "Number of TIER_0 entries",
"type": "integer"
},
"tier_1_count": {
"default": 0,
"description": "Number of TIER_1 entries",
"type": "integer"
},
"tier_2_count": {
"default": 0,
"description": "Number of TIER_2 entries",
"type": "integer"
},
"total": {
"default": 0,
"description": "Total number of entries",
"type": "integer"
}
},
"type": "object"
}
},
"required": [
"connector_version_id",
"connector_definition_id",
"docker_repository",
"days",
"customer_tier_filter",
"healthy",
"unhealthy",
"awaiting",
"disabled",
"total_actors",
"healthy_by_tier",
"unhealthy_by_tier",
"awaiting_by_tier",
"disabled_by_tier"
],
"type": "object"
}
query_prod_actors_by_pinned_connector_version
Hints: read-only · idempotent
List actors (sources/destinations) effectively pinned to a specific connector version.
Returns all actors that are effectively pinned to a specific connector version, considering all scope levels: actor-level pins, workspace-level pins, and organization-level pins (with actor > workspace > organization precedence). Useful for monitoring rollouts and understanding which customers are affected.
The actor_id field is the actor ID (superset of source_id/destination_id).
Returns list of dicts with keys: actor_id, connector_definition_id, origin_type, origin, description, created_at, expires_at, pin_scope_type, actor_name, workspace_id, workspace_name, organization_id, dataplane_group_id, dataplane_name
pin_scope_type is 'actor', 'workspace', or 'organization' indicating which scope level the effective pin came from.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_version_id |
string |
yes | — | Connector version UUID to find pinned instances for |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_version_id": {
"description": "Connector version UUID to find pinned instances for",
"type": "string"
}
},
"required": [
"connector_version_id"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_connection_sync_activity
Hints: read-only · idempotent · open-world
List recent sync jobs and attempts from the Prod DB Replica.
Returns one row per (job, attempt) pair for sync jobs whose updated_at
falls in [start_at, end_at), scoped to the provided organization,
workspace, or connection IDs. Designed for live operational lookups —
e.g. "what happened on this connection in the last hour" — not for
historical analysis.
Each row is enriched with customer_tier and is_eu for the owning
organization. Tier filtering is intentionally not applied — this is a
read-only observability query.
Input requirements:
- At least one of
organization_id,workspace_id, orconnection_idsmust be provided (any combination is accepted). start_atandend_atmust be timezone-aware andstart_at < end_at.
Key fields in each row:
job_id,attempt_id,attempt_numberjob_status,attempt_statusjob_started_at,job_updated_at,attempt_ended_atfailure_summary(JSON; populated when an attempt failed)connection_id,connection_name,connection_statussource_actor_id,source_actor_name,source_actor_definition_iddestination_actor_id,destination_actor_name,destination_actor_definition_idworkspace_id,workspace_name,organization_iddataplane_group_id,dataplane_namecustomer_tier,is_eu(added by tier enrichment)
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
start_at |
string |
yes | — | Inclusive start timestamp for the sync activity window. Must be timezone-aware (ISO 8601 with offset or Z). |
end_at |
string |
yes | — | Exclusive end timestamp for the sync activity window. Must be timezone-aware and strictly after start_at. |
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") | null |
no | null |
Optional organization UUID or alias. At least one of organization_id, workspace_id, or connection_ids is required. Accepts @airbyte-internal as an alias for the Airbyte internal org. |
workspace_id |
string | enum("266ebdfe-0d7b-4540-9817-de7e4505ba61") | null |
no | null |
Optional workspace UUID or alias. At least one of organization_id, workspace_id, or connection_ids is required. Accepts @devin-ai-sandbox as an alias for the Devin AI sandbox workspace. |
connection_ids |
array<string> | null |
no | null |
Optional list of connection UUIDs. At least one of organization_id, workspace_id, or connection_ids is required. |
status_filter |
enum("all", "succeeded", "failed") |
no | "all" |
Filter by job status: all (default), succeeded, or failed. Applied to jobs.status in the Prod DB Replica. |
limit |
integer |
no | 1000 |
Maximum number of attempt rows to return. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"start_at": {
"description": "Inclusive start timestamp for the sync activity window. Must be timezone-aware (ISO 8601 with offset or `Z`).",
"format": "date-time",
"type": "string"
},
"end_at": {
"description": "Exclusive end timestamp for the sync activity window. Must be timezone-aware and strictly after `start_at`.",
"format": "date-time",
"type": "string"
},
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional organization UUID or alias. At least one of `organization_id`, `workspace_id`, or `connection_ids` is required. Accepts `@airbyte-internal` as an alias for the Airbyte internal org."
},
"workspace_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Workspace ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@devin-ai-sandbox\") and its value\nis the actual workspace UUID. Use `WorkspaceAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"266ebdfe-0d7b-4540-9817-de7e4505ba61"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional workspace UUID or alias. At least one of `organization_id`, `workspace_id`, or `connection_ids` is required. Accepts `@devin-ai-sandbox` as an alias for the Devin AI sandbox workspace."
},
"connection_ids": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional list of connection UUIDs. At least one of `organization_id`, `workspace_id`, or `connection_ids` is required."
},
"status_filter": {
"description": "Filter by job status: `all` (default), `succeeded`, or `failed`. Applied to `jobs.status` in the Prod DB Replica.",
"enum": [
"all",
"succeeded",
"failed"
],
"type": "string",
"default": "all"
},
"limit": {
"default": 1000,
"description": "Maximum number of attempt rows to return.",
"type": "integer"
}
},
"required": [
"start_at",
"end_at"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_connections_by_connector
Hints: read-only · idempotent · open-world
Search for all connections using a specific source or destination connector type.
This tool queries the Airbyte Cloud Prod DB Replica directly for fast results. It finds all connections where the source or destination connector matches the specified type, regardless of how the connector is named by users.
Results are always enriched with customer_tier and is_eu fields. The customer_tier_filter parameter is required to ensure tier-aware querying.
Optionally filter by organization_id to limit results to a specific organization. Use '@airbyte-internal' as an alias for the Airbyte internal organization.
Set exclude_pinned=True to filter out connections that are already pinned to a
specific version. This is useful for 'prove fix' live connection testing workflows
where you want to find unpinned connections to test against.
Set enabled_schedules_only=True to restrict results to connections that are both
enabled (status='active') and on an automated schedule (not manual-trigger-only).
This is useful for canary prerelease workflows where you need connections that
will run organically during the monitoring window.
Returns a list of connection dicts with workspace context and clickable Cloud UI URLs. For source queries, returns: connection_id, connection_name, connection_url, source_id, source_name, source_definition_id, workspace_id, workspace_name, organization_id, dataplane_group_id, dataplane_name, pin_origin_type, pin_origin, pinned_version_id, pin_scope_type, customer_tier, is_eu. For destination queries, returns: connection_id, connection_name, connection_url, destination_id, destination_name, destination_definition_id, workspace_id, workspace_name, organization_id, dataplane_group_id, dataplane_name, pin_origin_type, pin_origin, pinned_version_id, pin_scope_type, customer_tier, is_eu.
pin_scope_type is 'actor', 'workspace', or 'organization' indicating which scope level the effective pin came from (NULL if not pinned).
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
source_definition_id |
string | null |
no | null |
Source connector definition ID (UUID) to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics. |
source_canonical_name |
string | null |
no | null |
Canonical source connector name to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Examples: 'source-youtube-analytics', 'YouTube Analytics'. |
destination_definition_id |
string | null |
no | null |
Destination connector definition ID (UUID) to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Example: 'e5c8e66c-a480-4a5e-9c0e-e8e5e4c5c5c5' for DuckDB. |
destination_canonical_name |
string | null |
no | null |
Canonical destination connector name to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Examples: 'destination-duckdb', 'DuckDB'. |
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") | null |
no | null |
Optional organization ID (UUID) or alias to filter results. If provided, only connections in this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org. |
limit |
integer |
no | 1000 |
Maximum number of results (default: 1000) |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers. |
exclude_pinned |
boolean |
no | false |
If True, exclude connections whose connector is already pinned to a specific version (at any scope level: actor, workspace, or organization). Useful for 'prove fix' workflows where you want to find unpinned connections for live testing. Default: False (include all connections). |
enabled_schedules_only |
boolean |
no | false |
If True, only return connections that are both active (not paused/inactive) and on an automated sync schedule (not manual-trigger-only). Useful for canary workflows where you need connections that will produce organic syncs during a monitoring window. Default: False (include all connections). |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"source_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Source connector definition ID (UUID) to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics."
},
"source_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical source connector name to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Examples: 'source-youtube-analytics', 'YouTube Analytics'."
},
"destination_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Destination connector definition ID (UUID) to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Example: 'e5c8e66c-a480-4a5e-9c0e-e8e5e4c5c5c5' for DuckDB."
},
"destination_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical destination connector name to search for. Exactly one of source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name is required. Examples: 'destination-duckdb', 'DuckDB'."
},
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional organization ID (UUID) or alias to filter results. If provided, only connections in this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org."
},
"limit": {
"default": 1000,
"description": "Maximum number of results (default: 1000)",
"type": "integer"
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
},
"exclude_pinned": {
"default": false,
"description": "If True, exclude connections whose connector is already pinned to a specific version (at any scope level: actor, workspace, or organization). Useful for 'prove fix' workflows where you want to find unpinned connections for live testing. Default: False (include all connections).",
"type": "boolean"
},
"enabled_schedules_only": {
"default": false,
"description": "If True, only return connections that are both active (not paused/inactive) and on an automated sync schedule (not manual-trigger-only). Useful for canary workflows where you need connections that will produce organic syncs during a monitoring window. Default: False (include all connections).",
"type": "boolean"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_connections_by_stream
Hints: read-only · idempotent · open-world
Find connections that have a specific stream enabled in their catalog.
This tool searches the connection's configured catalog (JSONB) for streams matching the specified name. It's particularly useful when validating connector fixes that affect specific streams - you can quickly find customer connections that use the affected stream.
Results are always enriched with customer_tier and is_eu fields. The customer_tier_filter parameter is required to ensure tier-aware querying.
Use cases:
- Finding connections with a specific stream enabled for regression testing
- Validating connector fixes that affect particular streams
- Identifying which customers use rarely-enabled streams
Returns a list of connection dicts with workspace context and clickable Cloud UI URLs.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
stream_name |
string |
yes | — | Name of the stream to search for in connection catalogs. This must match the exact stream name as configured in the connection. Examples: 'global_exclusions', 'campaigns', 'users'. |
source_definition_id |
string | null |
no | null |
Source connector definition ID (UUID) to search for. Provide this OR source_canonical_name (exactly one required). Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics. |
source_canonical_name |
string | null |
no | null |
Canonical source connector name to search for. Provide this OR source_definition_id (exactly one required). Examples: 'source-klaviyo', 'Klaviyo', 'source-youtube-analytics'. |
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") | null |
no | null |
Optional organization ID (UUID) or alias to filter results. If provided, only connections in this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org. |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"stream_name": {
"description": "Name of the stream to search for in connection catalogs. This must match the exact stream name as configured in the connection. Examples: 'global_exclusions', 'campaigns', 'users'.",
"type": "string"
},
"source_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Source connector definition ID (UUID) to search for. Provide this OR source_canonical_name (exactly one required). Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics."
},
"source_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical source connector name to search for. Provide this OR source_definition_id (exactly one required). Examples: 'source-klaviyo', 'Klaviyo', 'source-youtube-analytics'."
},
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional organization ID (UUID) or alias to filter results. If provided, only connections in this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org."
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
}
},
"required": [
"stream_name"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_connector_connection_stats
Hints: read-only · idempotent · open-world
Get aggregate connection stats for multiple connectors.
Returns counts of connections grouped by pinned version for each connector, including:
- Total, enabled, and active connection counts
- Pinned vs unpinned breakdown
- Latest attempt status breakdown (succeeded, failed, cancelled, running, unknown)
This tool is designed for release monitoring workflows. It allows you to:
- Query recently released connectors to identify which ones to monitor
- Get aggregate stats showing how many connections are using each version
- See health metrics (pass/fail) broken down by version
The lookback_days parameter controls the lookback window for:
- Counting 'active' connections (those with recent sync activity)
- Determining 'latest attempt status' (most recent attempt within the window)
Connections with no sync activity in the lookback window will have 'unknown' status in the latest_attempt breakdown.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
source_definition_ids |
array<string> | null |
no | null |
List of source connector definition IDs (UUIDs) to get stats for. Example: ['afa734e4-3571-11ec-991a-1e0031268139'] |
destination_definition_ids |
array<string> | null |
no | null |
List of destination connector definition IDs (UUIDs) to get stats for. Example: ['94bd199c-2ff0-4aa2-b98e-17f0acb72610'] |
lookback_days |
integer |
no | 7 |
Number of days to look back for 'active' connections (default: 7). Connections with sync activity within this window are counted as active. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"source_definition_ids": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"description": "List of source connector definition IDs (UUIDs) to get stats for. Example: ['afa734e4-3571-11ec-991a-1e0031268139']"
},
"destination_definition_ids": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"description": "List of destination connector definition IDs (UUIDs) to get stats for. Example: ['94bd199c-2ff0-4aa2-b98e-17f0acb72610']"
},
"lookback_days": {
"default": 7,
"description": "Number of days to look back for 'active' connections (default: 7). Connections with sync activity within this window are counted as active.",
"type": "integer"
}
},
"type": "object"
}
Show output JSON schema
{
"description": "Response containing connection stats for multiple connectors.",
"properties": {
"sources": {
"description": "Stats for source connectors",
"items": {
"description": "Aggregate connection stats for a connector.",
"properties": {
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_type": {
"description": "'source' or 'destination'",
"type": "string"
},
"canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The canonical connector name if resolved"
},
"total_connections": {
"description": "Total number of non-deprecated connections",
"type": "integer"
},
"enabled_connections": {
"description": "Number of enabled (active status) connections",
"type": "integer"
},
"active_connections": {
"description": "Number of connections with recent sync activity",
"type": "integer"
},
"pinned_connections": {
"description": "Number of connections with explicit version pins",
"type": "integer"
},
"unpinned_connections": {
"description": "Number of connections on default version",
"type": "integer"
},
"latest_attempt": {
"description": "Overall breakdown by latest attempt status",
"properties": {
"succeeded": {
"default": 0,
"description": "Connections where latest attempt succeeded",
"type": "integer"
},
"failed": {
"default": 0,
"description": "Connections where latest attempt failed",
"type": "integer"
},
"cancelled": {
"default": 0,
"description": "Connections where latest attempt was cancelled",
"type": "integer"
},
"running": {
"default": 0,
"description": "Connections where latest attempt is still running",
"type": "integer"
},
"unknown": {
"default": 0,
"description": "Connections with no recent attempts in the lookback window",
"type": "integer"
}
},
"type": "object"
},
"by_version": {
"description": "Stats broken down by pinned version",
"items": {
"description": "Stats for connections pinned to a specific version.",
"properties": {
"pinned_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "The connector version UUID (None for unpinned connections)"
},
"docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The docker image tag for this version"
},
"total_connections": {
"description": "Total number of connections",
"type": "integer"
},
"enabled_connections": {
"description": "Number of enabled (active status) connections",
"type": "integer"
},
"active_connections": {
"description": "Number of connections with recent sync activity",
"type": "integer"
},
"latest_attempt": {
"description": "Breakdown by latest attempt status",
"properties": {
"succeeded": {
"default": 0,
"description": "Connections where latest attempt succeeded",
"type": "integer"
},
"failed": {
"default": 0,
"description": "Connections where latest attempt failed",
"type": "integer"
},
"cancelled": {
"default": 0,
"description": "Connections where latest attempt was cancelled",
"type": "integer"
},
"running": {
"default": 0,
"description": "Connections where latest attempt is still running",
"type": "integer"
},
"unknown": {
"default": 0,
"description": "Connections with no recent attempts in the lookback window",
"type": "integer"
}
},
"type": "object"
}
},
"required": [
"pinned_version_id",
"total_connections",
"enabled_connections",
"active_connections",
"latest_attempt"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"connector_definition_id",
"connector_type",
"total_connections",
"enabled_connections",
"active_connections",
"pinned_connections",
"unpinned_connections",
"latest_attempt",
"by_version"
],
"type": "object"
},
"type": "array"
},
"destinations": {
"description": "Stats for destination connectors",
"items": {
"description": "Aggregate connection stats for a connector.",
"properties": {
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_type": {
"description": "'source' or 'destination'",
"type": "string"
},
"canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The canonical connector name if resolved"
},
"total_connections": {
"description": "Total number of non-deprecated connections",
"type": "integer"
},
"enabled_connections": {
"description": "Number of enabled (active status) connections",
"type": "integer"
},
"active_connections": {
"description": "Number of connections with recent sync activity",
"type": "integer"
},
"pinned_connections": {
"description": "Number of connections with explicit version pins",
"type": "integer"
},
"unpinned_connections": {
"description": "Number of connections on default version",
"type": "integer"
},
"latest_attempt": {
"description": "Overall breakdown by latest attempt status",
"properties": {
"succeeded": {
"default": 0,
"description": "Connections where latest attempt succeeded",
"type": "integer"
},
"failed": {
"default": 0,
"description": "Connections where latest attempt failed",
"type": "integer"
},
"cancelled": {
"default": 0,
"description": "Connections where latest attempt was cancelled",
"type": "integer"
},
"running": {
"default": 0,
"description": "Connections where latest attempt is still running",
"type": "integer"
},
"unknown": {
"default": 0,
"description": "Connections with no recent attempts in the lookback window",
"type": "integer"
}
},
"type": "object"
},
"by_version": {
"description": "Stats broken down by pinned version",
"items": {
"description": "Stats for connections pinned to a specific version.",
"properties": {
"pinned_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"description": "The connector version UUID (None for unpinned connections)"
},
"docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The docker image tag for this version"
},
"total_connections": {
"description": "Total number of connections",
"type": "integer"
},
"enabled_connections": {
"description": "Number of enabled (active status) connections",
"type": "integer"
},
"active_connections": {
"description": "Number of connections with recent sync activity",
"type": "integer"
},
"latest_attempt": {
"description": "Breakdown by latest attempt status",
"properties": {
"succeeded": {
"default": 0,
"description": "Connections where latest attempt succeeded",
"type": "integer"
},
"failed": {
"default": 0,
"description": "Connections where latest attempt failed",
"type": "integer"
},
"cancelled": {
"default": 0,
"description": "Connections where latest attempt was cancelled",
"type": "integer"
},
"running": {
"default": 0,
"description": "Connections where latest attempt is still running",
"type": "integer"
},
"unknown": {
"default": 0,
"description": "Connections with no recent attempts in the lookback window",
"type": "integer"
}
},
"type": "object"
}
},
"required": [
"pinned_version_id",
"total_connections",
"enabled_connections",
"active_connections",
"latest_attempt"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"connector_definition_id",
"connector_type",
"total_connections",
"enabled_connections",
"active_connections",
"pinned_connections",
"unpinned_connections",
"latest_attempt",
"by_version"
],
"type": "object"
},
"type": "array"
},
"lookback_days": {
"description": "Lookback window used for 'active' connections",
"type": "integer"
},
"generated_at": {
"description": "When this response was generated",
"format": "date-time",
"type": "string"
}
},
"required": [
"lookback_days",
"generated_at"
],
"type": "object"
}
query_prod_connector_rollouts
Hints: read-only · idempotent
Query connector rollouts with flexible filtering.
Returns rollouts based on the provided filters. If no filters are specified, returns all active rollouts. Useful for monitoring rollout status and history.
Filter behavior:
- rollout_id: Returns that specific rollout (ignores other filters)
- active_only: Returns only active (non-terminal) rollouts
- actor_definition_id: Returns rollouts for that specific connector
- No filters: Returns all active rollouts (same as active_only=True)
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
actor_definition_id |
string | null |
no | null |
Connector definition UUID to filter by (optional) |
rollout_id |
string | null |
no | null |
Specific rollout UUID to look up (optional) |
active_only |
boolean |
no | false |
If true, only return active (non-terminal) rollouts |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"actor_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector definition UUID to filter by (optional)"
},
"rollout_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Specific rollout UUID to look up (optional)"
},
"active_only": {
"default": false,
"description": "If true, only return active (non-terminal) rollouts",
"type": "boolean"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"description": "Information about a connector rollout.",
"properties": {
"rollout_id": {
"description": "The rollout UUID",
"type": "string"
},
"actor_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"state": {
"description": "Rollout state: initialized, workflow_started, in_progress, paused, finalizing, succeeded, errored, failed_rolled_back, canceled",
"type": "string"
},
"initial_rollout_pct": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"description": "Initial rollout percentage"
},
"current_target_rollout_pct": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"description": "Current target rollout percentage"
},
"final_target_rollout_pct": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"description": "Final target rollout percentage"
},
"has_breaking_changes": {
"description": "Whether the RC has breaking changes",
"type": "boolean"
},
"max_step_wait_time_mins": {
"anyOf": [
{
"type": "integer"
},
{
"type": "null"
}
],
"default": null,
"description": "Maximum wait time between rollout steps in minutes"
},
"rollout_strategy": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Rollout strategy: manual, automated, overridden"
},
"updated_by_user_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "User ID recorded as last updating the rollout"
},
"updated_by_user_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Name recorded as last updating the rollout"
},
"updated_by_user_email": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Email recorded as last updating the rollout"
},
"workflow_run_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Temporal workflow run ID"
},
"error_msg": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Error message if errored"
},
"failed_reason": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Reason for failure if failed"
},
"paused_reason": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Reason for pause if paused"
},
"tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional tag for the rollout"
},
"created_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "When the rollout was created"
},
"updated_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "When the rollout was last updated"
},
"completed_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "When the rollout completed (if terminal)"
},
"expires_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "When the rollout expires"
},
"rc_docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker image tag of the release candidate"
},
"rc_docker_repository": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker repository of the release candidate"
},
"initial_docker_image_tag": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker image tag of the initial version"
},
"initial_docker_repository": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Docker repository of the initial version"
},
"filters": {
"anyOf": [
{
"additionalProperties": true,
"type": "object"
},
{
"type": "null"
}
],
"default": null,
"description": "Raw rollout filters JSON (e.g., {'tierFilter': {'tier': 'TIER_0'}})"
},
"customer_tier": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Customer tier targeted by this rollout (extracted from filters), e.g., 'TIER_0', 'TIER_1'. None if no tier filter is set."
}
},
"required": [
"rollout_id",
"actor_definition_id",
"state",
"has_breaking_changes"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_connector_versions
Hints: read-only · idempotent
List all versions for a connector definition.
Returns all published versions of a connector, ordered by last_published date descending. Useful for understanding version history and finding specific version IDs for pinning or rollout monitoring.
Returns list of dicts with keys: version_id, docker_image_tag, docker_repository, release_stage, support_level, cdk_version, language, last_published, release_date
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_definition_id |
string |
yes | — | Connector definition UUID to list versions for |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_definition_id": {
"description": "Connector definition UUID to list versions for",
"type": "string"
}
},
"required": [
"connector_definition_id"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_dataplanes
Hints: read-only · idempotent
List all dataplane groups with workspace counts.
Returns information about all active dataplane groups in Airbyte Cloud, including the number of workspaces in each. Useful for understanding the distribution of workspaces across regions (US, US-Central, EU).
Returns list of dicts with keys: dataplane_group_id, dataplane_name, organization_id, enabled, tombstone, created_at, workspace_count
Parameters:
_No parameters._
Show input JSON schema
{
"additionalProperties": false,
"properties": {},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_failed_sync_attempts_for_connector
Hints: read-only · idempotent · open-world
List failed sync attempts for ALL actors using a source connector type.
This tool finds all actors with the given connector definition and returns their failed sync attempts, regardless of whether they have explicit version pins.
Results are always enriched with customer_tier and is_eu fields. The customer_tier_filter parameter is required to ensure tier-aware querying.
This is useful for investigating connector issues across all users. Use this when you want to find failures for a connector type regardless of which version users are on.
Note: This tool only supports SOURCE connectors. For destination connectors, a separate tool would be needed.
Key fields in results:
- failure_summary: JSON containing failure details including failureType and messages
- customer_tier: TIER_0, TIER_1, or TIER_2
- is_eu: Whether the workspace is in the EU region
- pin_origin_type, pin_origin, pinned_version_id: Version pin context (NULL if not pinned)
- pin_scope_type: 'actor', 'workspace', or 'organization' (NULL if not pinned)
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
source_definition_id |
string | null |
no | null |
Source connector definition ID (UUID) to search for. Exactly one of this or source_canonical_name is required. Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics. |
source_canonical_name |
string | null |
no | null |
Canonical source connector name to search for. Exactly one of this or source_definition_id is required. Examples: 'source-youtube-analytics', 'YouTube Analytics'. |
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") | null |
no | null |
Optional organization ID (UUID) or alias to filter results. If provided, only failed attempts from this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org. |
lookback_days |
integer |
no | 7 |
Number of days to look back (default: 7) |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"source_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Source connector definition ID (UUID) to search for. Exactly one of this or source_canonical_name is required. Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics."
},
"source_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical source connector name to search for. Exactly one of this or source_definition_id is required. Examples: 'source-youtube-analytics', 'YouTube Analytics'."
},
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional organization ID (UUID) or alias to filter results. If provided, only failed attempts from this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org."
},
"lookback_days": {
"default": 7,
"description": "Number of days to look back (default: 7)",
"type": "integer"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_new_connector_releases
Hints: read-only · idempotent
List recently published connector versions.
Returns connector versions published within the specified number of days. Uses last_published timestamp which reflects when the version was actually deployed to the registry (not the changelog date).
Returns list of dicts with keys: version_id, connector_definition_id, docker_repository, docker_image_tag, last_published, release_date, release_stage, support_level, cdk_version, language, created_at
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
days |
integer |
no | 7 |
Number of days to look back (default: 7) |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"days": {
"default": 7,
"description": "Number of days to look back (default: 7)",
"type": "integer"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_organizations
Hints: read-only · idempotent
Search organizations by name or email substring.
Performs a case-insensitive substring match on organization name and email.
Use the returned organization_id values with other tools like
query_prod_connections_by_connector or lookup_customer_tiers.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
name_contains |
string |
yes | — | Case-insensitive substring to search for in organization name or email. For example, 'acme' will match organizations named 'Acme Corp' or with email 'admin@acme.io'. |
limit |
integer |
no | 20 |
Maximum number of organizations to return (default: 20) |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"name_contains": {
"description": "Case-insensitive substring to search for in organization name or email. For example, 'acme' will match organizations named 'Acme Corp' or with email 'admin@acme.io'.",
"type": "string"
},
"limit": {
"default": 20,
"description": "Maximum number of organizations to return (default: 20)",
"type": "integer"
}
},
"required": [
"name_contains"
],
"type": "object"
}
Show output JSON schema
{
"description": "Result of searching organizations by name substring.",
"properties": {
"name_contains": {
"description": "The search substring that was used",
"type": "string"
},
"total_found": {
"description": "Total number of organizations matching",
"type": "integer"
},
"organizations": {
"description": "List of matching organizations",
"items": {
"description": "A single organization returned by a name/email search.",
"properties": {
"organization_id": {
"description": "The organization UUID",
"type": "string"
},
"organization_name": {
"description": "The name of the organization",
"type": "string"
},
"email": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The email address associated with the organization"
},
"customer_tier": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Customer tier (TIER_0, TIER_1, or TIER_2). Enriched from the GCS tier cache."
}
},
"required": [
"organization_id",
"organization_name"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"name_contains",
"total_found",
"organizations"
],
"type": "object"
}
query_prod_pin_stats_for_organization
Hints: read-only · idempotent · open-world
Query connector versions pinned anywhere under an organization.
Returns one row per pinned version, aggregating every connector_version
pin whose scope belongs to the organization — the org itself, one of its
workspaces, or an actor within one of those workspaces (actor, workspace,
and organization scopes). Each row carries the per-scope pin breakdown, the
manual/rollout/breaking-change split, and a has_active_rollout flag.
This powers the first step of the Organization Pins view (pick an org, then
see the versions pinned under it). Use query_prod_pins_for_organization
for the individual pins behind a selected version.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") |
yes | — | Organization UUID (or @airbyte-internal alias) to scope pins to. Resolve organization names to an ID first via search_organizations. |
connector_definition_id |
string | null |
no | null |
Connector definition UUID to filter by (optional). Mutually exclusive with connector_canonical_name. |
connector_canonical_name |
string | null |
no | null |
Connector canonical name (e.g. source-postgres) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with connector_definition_id. |
limit |
integer |
no | 1000 |
Maximum number of versions to return (default: 1000). |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
}
],
"description": "Organization UUID (or `@airbyte-internal` alias) to scope pins to. Resolve organization names to an ID first via `search_organizations`."
},
"connector_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector definition UUID to filter by (optional). Mutually exclusive with `connector_canonical_name`."
},
"connector_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector canonical name (e.g. `source-postgres`) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with `connector_definition_id`."
},
"limit": {
"default": 1000,
"description": "Maximum number of versions to return (default: 1000).",
"type": "integer"
}
},
"required": [
"organization_id"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"description": "A connector version pinned somewhere under an organization, with counts.",
"properties": {
"version_id": {
"description": "The actor_definition_version UUID",
"type": "string"
},
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_name": {
"description": "Human-readable connector name",
"type": "string"
},
"docker_repository": {
"description": "Docker repository path",
"type": "string"
},
"docker_image_tag": {
"description": "Docker image tag for this version",
"type": "string"
},
"last_published": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "ISO timestamp when this version was last published"
},
"pin_count": {
"description": "Total pins under the org targeting this version (all scopes)",
"type": "integer"
},
"manual_pins": {
"default": 0,
"description": "Pins with no system origin (user-created manual pins), any scope",
"type": "integer"
},
"rollout_pins": {
"default": 0,
"description": "Pins created by connector rollouts",
"type": "integer"
},
"breaking_change_pins": {
"default": 0,
"description": "Pins created by breaking changes",
"type": "integer"
},
"actor_pins": {
"description": "Manual actor-scoped pins (excludes rollout and breaking-change)",
"type": "integer"
},
"workspace_pins": {
"description": "Workspace-scoped pins under the org",
"type": "integer"
},
"org_pins": {
"description": "Organization-scoped pins",
"type": "integer"
},
"has_active_rollout": {
"default": false,
"description": "`True` if at least one rollout pin is backed by a non-terminal `connector_rollout`",
"type": "boolean"
}
},
"required": [
"version_id",
"connector_definition_id",
"connector_name",
"docker_repository",
"docker_image_tag",
"pin_count",
"actor_pins",
"workspace_pins",
"org_pins"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_pins_for_organization
Hints: read-only · idempotent · open-world
List the individual connector-version pins discovered under an organization.
Returns one row per scoped_configuration pin whose scope belongs to the
organization (org/workspace/actor), resolving the pinned connector and
version, the scope's display name, the manual author's email, and — for
rollout-origin pins — the backing connector_rollout id and state. This
directly answers whether each pin is manual or caused by an active rollout.
This powers the second step of the Organization Pins view: after picking a
version from query_prod_pin_stats_for_organization, pass its
pinned_version_id here to list the pins behind it.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") |
yes | — | Organization UUID (or @airbyte-internal alias) to scope pins to. Resolve organization names to an ID first via search_organizations. |
connector_definition_id |
string | null |
no | null |
Connector definition UUID to filter by (optional). Mutually exclusive with connector_canonical_name. |
connector_canonical_name |
string | null |
no | null |
Connector canonical name (e.g. source-postgres) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with connector_definition_id. |
pinned_version_id |
string | null |
no | null |
Actor_definition_version UUID to return only pins targeting that version. This is the post-selection filter for the org pins tab. |
origin_filter |
enum("all", "manual", "connector_rollout", "breaking_change") |
no | "all" |
Restrict by how the pin was created: all (default), manual, connector_rollout, or breaking_change. |
limit |
integer |
no | 1000 |
Maximum number of pins to return (default: 1000). |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
}
],
"description": "Organization UUID (or `@airbyte-internal` alias) to scope pins to. Resolve organization names to an ID first via `search_organizations`."
},
"connector_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector definition UUID to filter by (optional). Mutually exclusive with `connector_canonical_name`."
},
"connector_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector canonical name (e.g. `source-postgres`) to filter by. Resolved to a definition ID via the registry. Mutually exclusive with `connector_definition_id`."
},
"pinned_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Actor_definition_version UUID to return only pins targeting that version. This is the post-selection filter for the org pins tab."
},
"origin_filter": {
"description": "Restrict by how the pin was created: `all` (default), `manual`, `connector_rollout`, or `breaking_change`.",
"enum": [
"all",
"manual",
"connector_rollout",
"breaking_change"
],
"type": "string",
"default": "all"
},
"limit": {
"default": 1000,
"description": "Maximum number of pins to return (default: 1000).",
"type": "integer"
}
},
"required": [
"organization_id"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"description": "A single `scoped_configuration` pin discovered under an organization.",
"properties": {
"connector_definition_id": {
"description": "The connector definition UUID",
"type": "string"
},
"connector_name": {
"description": "Human-readable connector name",
"type": "string"
},
"docker_repository": {
"description": "Docker repository path",
"type": "string"
},
"pinned_version_id": {
"description": "The pinned actor_definition_version UUID",
"type": "string"
},
"pinned_version_tag": {
"description": "Docker image tag of the pinned version",
"type": "string"
},
"pin_scope_type": {
"description": "Scope of the pin: `organization`, `workspace`, or `actor`",
"type": "string"
},
"scope_id": {
"description": "UUID of the scoped entity",
"type": "string"
},
"scope_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Display name of the scoped entity, when resolvable"
},
"pin_category": {
"description": "Derived pin type: `manual`, `rollout`, or `breaking_change`",
"type": "string"
},
"set_by": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Email (or name) of the user who set a manual pin, when known"
},
"rollout_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Backing connector_rollout UUID for rollout pins"
},
"rollout_state": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "State of the backing rollout, for rollout pins"
},
"is_active_rollout": {
"default": false,
"description": "`True` when `rollout_state` is a non-terminal (active) state",
"type": "boolean"
},
"description": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Free-text pin reason"
},
"reference_url": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Reference URL attached to the pin, when present"
},
"created_at": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "ISO timestamp when the pin was created"
},
"expires_at": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "ISO timestamp when the pin expires, when set"
}
},
"required": [
"connector_definition_id",
"connector_name",
"docker_repository",
"pinned_version_id",
"pinned_version_tag",
"pin_scope_type",
"scope_id",
"pin_category"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_recent_syncs_for_connector
Hints: read-only · idempotent · open-world
List recent sync jobs for ALL actors using a connector type.
This tool finds all actors with the given connector definition and returns their recent sync jobs, regardless of whether they have explicit version pins. It filters out deleted actors, deleted workspaces, and deprecated connections.
Results are always enriched with customer_tier and is_eu fields. The customer_tier_filter parameter is required to ensure tier-aware querying.
Use this tool to:
- Find healthy connections with recent successful syncs (status_filter='succeeded')
- Investigate connector issues across all users (status_filter='failed')
- Get an overview of all recent sync activity (status_filter='all')
Set exclude_pinned=True to filter out syncs for actors that are already pinned to a
specific version. This is useful for 'prove fix' live connection testing workflows
where you want to find unpinned connections to test against.
Set enabled_schedules_only=True to restrict results to connections that are both
enabled (status='active') and on an automated schedule (not manual-trigger-only).
This is useful for canary prerelease workflows where you need connections that
will run organically during the monitoring window.
Supports both SOURCE and DESTINATION connectors. Provide exactly one of: source_definition_id, source_canonical_name, destination_definition_id, or destination_canonical_name.
Key fields in results:
- job_status: 'succeeded', 'failed', 'cancelled', etc.
- connection_id, connection_name: The connection that ran the sync
- actor_id, actor_name: The source or destination actor
- customer_tier: TIER_0, TIER_1, or TIER_2
- is_eu: Whether the workspace is in the EU region
- pin_origin_type, pin_origin, pinned_version_id: Version pin context (NULL if not pinned)
- pin_scope_type: 'actor', 'workspace', or 'organization' (NULL if not pinned)
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
source_definition_id |
string | null |
no | null |
Source connector definition ID (UUID) to search for. Provide this OR source_canonical_name OR destination_definition_id OR destination_canonical_name (exactly one required). Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics. |
source_canonical_name |
string | null |
no | null |
Canonical source connector name to search for. Provide this OR source_definition_id OR destination_definition_id OR destination_canonical_name (exactly one required). Examples: 'source-youtube-analytics', 'YouTube Analytics'. |
destination_definition_id |
string | null |
no | null |
Destination connector definition ID (UUID) to search for. Provide this OR destination_canonical_name OR source_definition_id OR source_canonical_name (exactly one required). Example: '94bd199c-2ff0-4aa2-b98e-17f0acb72610' for DuckDB. |
destination_canonical_name |
string | null |
no | null |
Canonical destination connector name to search for. Provide this OR destination_definition_id OR source_definition_id OR source_canonical_name (exactly one required). Examples: 'destination-duckdb', 'DuckDB'. |
status_filter |
enum("all", "succeeded", "failed") |
no | "all" |
Filter by job status: 'all' (default), 'succeeded', or 'failed'. Use 'succeeded' to find healthy connections with recent successful syncs. Use 'failed' to find connections with recent failures. |
organization_id |
string | enum("664c690e-5263-49ba-b01f-4a6759b3330a") | null |
no | null |
Optional organization ID (UUID) or alias to filter results. If provided, only syncs from this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org. |
lookback_days |
integer |
no | 7 |
Number of days to look back (default: 7) |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
customer_tier_filter |
enum("TIER_0", "TIER_1", "TIER_2", "ALL") |
no | "TIER_2" |
Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers. |
exclude_pinned |
boolean |
no | false |
If True, exclude syncs for actors that are already pinned to a specific version (at any scope level: actor, workspace, or organization). Useful for 'prove fix' workflows where you want to find unpinned connections for live testing. Default: False (include all syncs). |
enabled_schedules_only |
boolean |
no | false |
If True, only return syncs for connections that are both active (not paused/inactive) and on an automated sync schedule (not manual-trigger-only). Useful for canary workflows where you need connections that will produce organic syncs during a monitoring window. Default: False (include all connections). |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"source_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Source connector definition ID (UUID) to search for. Provide this OR source_canonical_name OR destination_definition_id OR destination_canonical_name (exactly one required). Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics."
},
"source_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical source connector name to search for. Provide this OR source_definition_id OR destination_definition_id OR destination_canonical_name (exactly one required). Examples: 'source-youtube-analytics', 'YouTube Analytics'."
},
"destination_definition_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Destination connector definition ID (UUID) to search for. Provide this OR destination_canonical_name OR source_definition_id OR source_canonical_name (exactly one required). Example: '94bd199c-2ff0-4aa2-b98e-17f0acb72610' for DuckDB."
},
"destination_canonical_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical destination connector name to search for. Provide this OR destination_definition_id OR source_definition_id OR source_canonical_name (exactly one required). Examples: 'destination-duckdb', 'DuckDB'."
},
"status_filter": {
"description": "Filter by job status: 'all' (default), 'succeeded', or 'failed'. Use 'succeeded' to find healthy connections with recent successful syncs. Use 'failed' to find connections with recent failures.",
"enum": [
"all",
"succeeded",
"failed"
],
"type": "string",
"default": "all"
},
"organization_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Organization ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@airbyte-internal\") and its value\nis the actual organization UUID. Use `OrganizationAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"664c690e-5263-49ba-b01f-4a6759b3330a"
],
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Optional organization ID (UUID) or alias to filter results. If provided, only syncs from this organization will be returned. Accepts '@airbyte-internal' as an alias for the Airbyte internal org."
},
"lookback_days": {
"default": 7,
"description": "Number of days to look back (default: 7)",
"type": "integer"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
},
"customer_tier_filter": {
"default": "TIER_2",
"description": "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. Filters results to only include connections belonging to organizations in the specified tier. Use 'ALL' to include all tiers.",
"enum": [
"TIER_0",
"TIER_1",
"TIER_2",
"ALL"
],
"type": "string"
},
"exclude_pinned": {
"default": false,
"description": "If True, exclude syncs for actors that are already pinned to a specific version (at any scope level: actor, workspace, or organization). Useful for 'prove fix' workflows where you want to find unpinned connections for live testing. Default: False (include all syncs).",
"type": "boolean"
},
"enabled_schedules_only": {
"default": false,
"description": "If True, only return syncs for connections that are both active (not paused/inactive) and on an automated sync schedule (not manual-trigger-only). Useful for canary workflows where you need connections that will produce organic syncs during a monitoring window. Default: False (include all connections).",
"type": "boolean"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_recent_syncs_for_connector_version
Hints: read-only · idempotent
List sync jobs that were run with a specific connector version.
Works for both source and destination connectors. Automatically detects the connector type from the version metadata and uses the appropriate query variant.
Accepts either connector_version_id (UUID) or connector_name +
connector_version (e.g. source-pokeapi + 0.3.59). When using
name + version, the docker_repository is derived from the canonical
name (e.g. source-pokeapi → airbyte/source-pokeapi).
Filters on the version stamped into jobs.config at job-creation time,
not the current pin state. This avoids false positives (pre-pin syncs
counted as RC) and false negatives (post-unpin syncs missed).
Pin columns (pin_origin_type, pin_origin, pin_scope_type) are
still included as informational output but are not used for filtering.
Returns list of dicts with keys: job_id, connection_id, job_status,
started_at, job_updated_at, connection_name, actor_id, actor_name,
actor_definition_id, source_definition_version_id,
destination_definition_version_id, pin_origin_type,
pin_origin, pin_scope_type, workspace_id, workspace_name,
organization_id, dataplane_group_id, dataplane_name.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
connector_version_id |
string | null |
no | null |
Connector version UUID. Provide this OR connector_name + connector_version. |
connector_name |
string | null |
no | null |
Canonical connector name (e.g. source-pokeapi, destination-duckdb). Used with connector_version to resolve the version UUID. |
connector_version |
string | null |
no | null |
Semver version tag (e.g. 0.3.59). Used with connector_name to resolve the version UUID. |
days |
integer |
no | 7 |
Number of days to look back (default: 7) |
limit |
integer |
no | 100 |
Maximum number of results (default: 100) |
successful_only |
boolean |
no | false |
If True, only return successful syncs (default: False) |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"connector_version_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Connector version UUID. Provide this OR connector_name + connector_version."
},
"connector_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Canonical connector name (e.g. `source-pokeapi`, `destination-duckdb`). Used with `connector_version` to resolve the version UUID."
},
"connector_version": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Semver version tag (e.g. `0.3.59`). Used with `connector_name` to resolve the version UUID."
},
"days": {
"default": 7,
"description": "Number of days to look back (default: 7)",
"type": "integer"
},
"limit": {
"default": 100,
"description": "Maximum number of results (default: 100)",
"type": "integer"
},
"successful_only": {
"default": false,
"description": "If `True`, only return successful syncs (default: `False`)",
"type": "boolean"
}
},
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"items": {
"additionalProperties": true,
"type": "object"
},
"type": "array"
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_workspace_info
Hints: read-only · idempotent
Get workspace information including dataplane group.
Returns details about a specific workspace, including which dataplane (region) it belongs to. Useful for determining if a workspace is in the EU region for filtering purposes.
Returns dict with keys: workspace_id, workspace_name, slug, organization_id, dataplane_group_id, dataplane_name, created_at, tombstone Or None if workspace not found.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
workspace_id |
string | enum("266ebdfe-0d7b-4540-9817-de7e4505ba61") |
yes | — | Workspace UUID or alias to look up. Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace. |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"workspace_id": {
"anyOf": [
{
"type": "string"
},
{
"description": "Workspace ID aliases that can be used in place of UUIDs.\n\nEach member's name is the alias (e.g., \"@devin-ai-sandbox\") and its value\nis the actual workspace UUID. Use `WorkspaceAliasEnum.resolve()` to\nresolve aliases to actual IDs.",
"enum": [
"266ebdfe-0d7b-4540-9817-de7e4505ba61"
],
"type": "string"
}
],
"description": "Workspace UUID or alias to look up. Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace."
}
},
"required": [
"workspace_id"
],
"type": "object"
}
Show output JSON schema
{
"properties": {
"result": {
"anyOf": [
{
"additionalProperties": true,
"type": "object"
},
{
"type": "null"
}
]
}
},
"required": [
"result"
],
"type": "object",
"x-fastmcp-wrap-result": true
}
query_prod_workspaces
Hints: read-only · idempotent
Search workspaces by name substring or email domain.
At least one of name_contains or email_domain must be provided.
When name_contains is given, performs a case-insensitive substring match
on workspace name and slug. When email_domain is given, matches
workspaces by user email domain.
The returned organization IDs can be used with other tools like
query_prod_connections_by_connector to find connections within
those organizations for safe testing.
Parameters:
| Name | Type | Required | Default | Description |
|---|---|---|---|---|
name_contains |
string | null |
no | null |
Case-insensitive substring to search for in workspace name or slug. For example, 'acme' will match workspaces named 'Acme Staging'. |
email_domain |
string | null |
no | null |
Email domain to search for (e.g., 'motherduck.com'). Do not include the '@' symbol. |
limit |
integer |
no | 100 |
Maximum number of workspaces to return (default: 100) |
Show input JSON schema
{
"additionalProperties": false,
"properties": {
"name_contains": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Case-insensitive substring to search for in workspace name or slug. For example, 'acme' will match workspaces named 'Acme Staging'."
},
"email_domain": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Email domain to search for (e.g., 'motherduck.com'). Do not include the '@' symbol."
},
"limit": {
"default": 100,
"description": "Maximum number of workspaces to return (default: 100)",
"type": "integer"
}
},
"type": "object"
}
Show output JSON schema
{
"description": "Result of searching workspaces by name or email domain.",
"properties": {
"name_contains": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The name substring that was searched for"
},
"email_domain": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The email domain that was searched for (e.g., 'motherduck.com')"
},
"total_workspaces_found": {
"description": "Total number of workspaces matching",
"type": "integer"
},
"unique_organization_ids": {
"description": "List of unique organization IDs found",
"items": {
"type": "string"
},
"type": "array"
},
"workspaces": {
"description": "List of matching workspaces",
"items": {
"description": "Information about a workspace.",
"properties": {
"organization_id": {
"description": "The organization UUID",
"type": "string"
},
"workspace_id": {
"description": "The workspace UUID",
"type": "string"
},
"workspace_name": {
"description": "The name of the workspace",
"type": "string"
},
"slug": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The workspace slug (URL-friendly identifier)"
},
"email": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The email address associated with the workspace"
},
"dataplane_group_id": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The dataplane group UUID (region)"
},
"dataplane_name": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "The name of the dataplane (e.g., 'US', 'EU')"
},
"created_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "When the workspace was created"
},
"customer_tier": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Customer tier (TIER_0, TIER_1, or TIER_2). Enriched from the GCS tier cache."
},
"is_eu": {
"anyOf": [
{
"type": "boolean"
},
{
"type": "null"
}
],
"default": null,
"description": "Whether the workspace is in the EU region (derived from dataplane_name)."
}
},
"required": [
"organization_id",
"workspace_id",
"workspace_name"
],
"type": "object"
},
"type": "array"
}
},
"required": [
"total_workspaces_found",
"unique_organization_ids",
"workspaces"
],
"type": "object"
}
1# Copyright (c) 2025 Airbyte, Inc., all rights reserved. 2"""MCP tools for querying the Airbyte Cloud Prod DB Replica. 3 4This module provides MCP tools that wrap the query functions from 5airbyte_ops_mcp.prod_db_access.queries for use by AI agents. 6 7## MCP reference 8 9.. include:: ../../../docs/mcp-generated/prod_db_ops.md 10 :start-line: 2 11""" 12 13from __future__ import annotations 14 15__all__: list[str] = [] 16 17import json 18import uuid 19from datetime import datetime, timezone 20from enum import StrEnum 21from typing import Annotated, Any 22 23from airbyte.exceptions import PyAirbyteInputError 24from fastmcp import FastMCP 25from fastmcp_extensions import mcp_tool, register_mcp_tools 26from pydantic import BaseModel, Field 27 28from airbyte_ops_mcp.cloud_admin.registry_lookup import ( 29 resolve_canonical_name_to_definition_id, 30) 31from airbyte_ops_mcp.constants import OrganizationAliasEnum, WorkspaceAliasEnum 32from airbyte_ops_mcp.prod_db_access.queries import ( 33 is_source_connector, 34 query_actor_population_by_org, 35 query_actors_pinned_to_version, 36 query_connection_sync_activity_from_prod, 37 query_connections_by_connector, 38 query_connections_by_destination_connector, 39 query_connections_by_stream, 40 query_connector_rollouts, 41 query_connector_versions, 42 query_dataplanes_list, 43 query_destination_connection_stats, 44 query_failed_sync_attempts_for_connector, 45 query_new_connector_releases, 46 query_org_connector_pins, 47 query_org_pin_stats, 48 query_recent_syncs_for_connector, 49 query_source_connection_stats, 50 query_syncs_for_connector_version, 51 query_version_actor_health, 52 query_versions_with_pins, 53 query_workspace_info, 54 query_workspaces_by_email_domain, 55 resolve_version_id_by_tag, 56 resolve_version_info, 57 search_organizations, 58 search_workspaces, 59) 60from airbyte_ops_mcp.tier_cache import ( 61 TierFilter, 62 TierSummary, 63 enrich_rows_by_org, 64 filter_rows_by_tier, 65 get_org_tiers, 66) 67from airbyte_ops_mcp.version_summaries import ( 68 summarize_population, 69 summarize_version_health, 70) 71 72 73class StatusFilter(StrEnum): 74 """Filter for job status in sync queries.""" 75 76 ALL = "all" 77 SUCCEEDED = "succeeded" 78 FAILED = "failed" 79 80 81# Cloud UI base URL for building connection URLs 82CLOUD_UI_BASE_URL = "https://cloud.airbyte.com" 83 84 85def _validate_sync_activity_scope( 86 *, 87 organization_id: str | None, 88 workspace_id: str | None, 89 connection_ids: list[str] | None, 90) -> None: 91 """Require at least one explicit scope filter for sync activity queries.""" 92 if organization_id or workspace_id or connection_ids: 93 return 94 raise PyAirbyteInputError( 95 message=( 96 "Provide at least one scope filter: `organization_id`, `workspace_id`, " 97 "or `connection_ids`." 98 ), 99 context={ 100 "organization_id": organization_id, 101 "workspace_id": workspace_id, 102 "connection_ids": connection_ids, 103 }, 104 ) 105 106 107def _validate_sync_activity_window( 108 *, 109 start_at: datetime, 110 end_at: datetime, 111) -> tuple[datetime, datetime]: 112 """Validate that `start_at` and `end_at` describe a usable window. 113 114 Returns the timestamps normalized to UTC. Raises `PyAirbyteInputError` for 115 naive timestamps or inverted ranges. No clock-relative caps are enforced 116 here; the caller is trusted to choose a sensible window. 117 """ 118 if start_at.tzinfo is None or end_at.tzinfo is None: 119 raise PyAirbyteInputError( 120 message="`start_at` and `end_at` must include timezone information.", 121 context={ 122 "start_at": start_at.isoformat(), 123 "end_at": end_at.isoformat(), 124 }, 125 ) 126 127 normalized_start = start_at.astimezone(timezone.utc) 128 normalized_end = end_at.astimezone(timezone.utc) 129 130 if normalized_start >= normalized_end: 131 raise PyAirbyteInputError( 132 message="`start_at` must be earlier than `end_at`.", 133 context={ 134 "start_at": normalized_start.isoformat(), 135 "end_at": normalized_end.isoformat(), 136 }, 137 ) 138 return normalized_start, normalized_end 139 140 141# ============================================================================= 142# Pydantic Models for MCP Tool Responses 143# ============================================================================= 144 145 146class OrganizationSearchHit(BaseModel): 147 """A single organization returned by a name/email search.""" 148 149 organization_id: str = Field(description="The organization UUID") 150 organization_name: str = Field(description="The name of the organization") 151 email: str | None = Field( 152 default=None, description="The email address associated with the organization" 153 ) 154 customer_tier: str | None = Field( 155 default=None, 156 description="Customer tier (TIER_0, TIER_1, or TIER_2). Enriched from the GCS tier cache.", 157 ) 158 159 160class OrganizationSearchResult(BaseModel): 161 """Result of searching organizations by name substring.""" 162 163 name_contains: str = Field(description="The search substring that was used") 164 total_found: int = Field(description="Total number of organizations matching") 165 organizations: list[OrganizationSearchHit] = Field( 166 description="List of matching organizations" 167 ) 168 169 170class WorkspaceInfo(BaseModel): 171 """Information about a workspace.""" 172 173 organization_id: str = Field(description="The organization UUID") 174 workspace_id: str = Field(description="The workspace UUID") 175 workspace_name: str = Field(description="The name of the workspace") 176 slug: str | None = Field( 177 default=None, description="The workspace slug (URL-friendly identifier)" 178 ) 179 email: str | None = Field( 180 default=None, description="The email address associated with the workspace" 181 ) 182 dataplane_group_id: str | None = Field( 183 default=None, description="The dataplane group UUID (region)" 184 ) 185 dataplane_name: str | None = Field( 186 default=None, description="The name of the dataplane (e.g., 'US', 'EU')" 187 ) 188 created_at: datetime | None = Field( 189 default=None, description="When the workspace was created" 190 ) 191 customer_tier: str | None = Field( 192 default=None, 193 description="Customer tier (TIER_0, TIER_1, or TIER_2). Enriched from the GCS tier cache.", 194 ) 195 is_eu: bool | None = Field( 196 default=None, 197 description="Whether the workspace is in the EU region (derived from dataplane_name).", 198 ) 199 200 201class WorkspaceSearchResult(BaseModel): 202 """Result of searching workspaces by name or email domain.""" 203 204 name_contains: str | None = Field( 205 default=None, description="The name substring that was searched for" 206 ) 207 email_domain: str | None = Field( 208 default=None, 209 description="The email domain that was searched for (e.g., 'motherduck.com')", 210 ) 211 total_workspaces_found: int = Field( 212 description="Total number of workspaces matching" 213 ) 214 unique_organization_ids: list[str] = Field( 215 description="List of unique organization IDs found" 216 ) 217 workspaces: list[WorkspaceInfo] = Field(description="List of matching workspaces") 218 219 220# Keep backward-compatible alias for any external references 221WorkspacesByEmailDomainResult = WorkspaceSearchResult 222 223 224class LatestAttemptBreakdown(BaseModel): 225 """Breakdown of connections by latest attempt status.""" 226 227 succeeded: int = Field( 228 default=0, description="Connections where latest attempt succeeded" 229 ) 230 failed: int = Field( 231 default=0, description="Connections where latest attempt failed" 232 ) 233 cancelled: int = Field( 234 default=0, description="Connections where latest attempt was cancelled" 235 ) 236 running: int = Field( 237 default=0, description="Connections where latest attempt is still running" 238 ) 239 unknown: int = Field( 240 default=0, 241 description="Connections with no recent attempts in the lookback window", 242 ) 243 244 245class VersionPinStats(BaseModel): 246 """Stats for connections pinned to a specific version.""" 247 248 pinned_version_id: str | None = Field( 249 description="The connector version UUID (None for unpinned connections)" 250 ) 251 docker_image_tag: str | None = Field( 252 default=None, description="The docker image tag for this version" 253 ) 254 total_connections: int = Field(description="Total number of connections") 255 enabled_connections: int = Field( 256 description="Number of enabled (active status) connections" 257 ) 258 active_connections: int = Field( 259 description="Number of connections with recent sync activity" 260 ) 261 latest_attempt: LatestAttemptBreakdown = Field( 262 description="Breakdown by latest attempt status" 263 ) 264 265 266class ConnectorConnectionStats(BaseModel): 267 """Aggregate connection stats for a connector.""" 268 269 connector_definition_id: str = Field(description="The connector definition UUID") 270 connector_type: str = Field(description="'source' or 'destination'") 271 canonical_name: str | None = Field( 272 default=None, description="The canonical connector name if resolved" 273 ) 274 total_connections: int = Field( 275 description="Total number of non-deprecated connections" 276 ) 277 enabled_connections: int = Field( 278 description="Number of enabled (active status) connections" 279 ) 280 active_connections: int = Field( 281 description="Number of connections with recent sync activity" 282 ) 283 pinned_connections: int = Field( 284 description="Number of connections with explicit version pins" 285 ) 286 unpinned_connections: int = Field( 287 description="Number of connections on default version" 288 ) 289 latest_attempt: LatestAttemptBreakdown = Field( 290 description="Overall breakdown by latest attempt status" 291 ) 292 by_version: list[VersionPinStats] = Field( 293 description="Stats broken down by pinned version" 294 ) 295 296 297class ConnectorConnectionStatsResponse(BaseModel): 298 """Response containing connection stats for multiple connectors.""" 299 300 sources: list[ConnectorConnectionStats] = Field( 301 default_factory=list, description="Stats for source connectors" 302 ) 303 destinations: list[ConnectorConnectionStats] = Field( 304 default_factory=list, description="Stats for destination connectors" 305 ) 306 lookback_days: int = Field( 307 description="Lookback window used for 'active' connections" 308 ) 309 generated_at: datetime = Field(description="When this response was generated") 310 311 312def _opt_str(value: Any) -> str | None: 313 """Convert a nullable value to str, returning None if the value is None/falsy.""" 314 return str(value) if value else None 315 316 317@mcp_tool( 318 read_only=True, 319 idempotent=True, 320) 321def query_prod_dataplanes() -> list[dict[str, Any]]: 322 """List all dataplane groups with workspace counts. 323 324 Returns information about all active dataplane groups in Airbyte Cloud, 325 including the number of workspaces in each. Useful for understanding 326 the distribution of workspaces across regions (US, US-Central, EU). 327 328 Returns list of dicts with keys: dataplane_group_id, dataplane_name, 329 organization_id, enabled, tombstone, created_at, workspace_count 330 """ 331 return query_dataplanes_list() 332 333 334@mcp_tool( 335 read_only=True, 336 idempotent=True, 337) 338def query_prod_workspace_info( 339 workspace_id: Annotated[ 340 str | WorkspaceAliasEnum, 341 Field( 342 description="Workspace UUID or alias to look up. " 343 "Accepts '@devin-ai-sandbox' as an alias for the Devin AI sandbox workspace." 344 ), 345 ], 346) -> dict[str, Any] | None: 347 """Get workspace information including dataplane group. 348 349 Returns details about a specific workspace, including which dataplane 350 (region) it belongs to. Useful for determining if a workspace is in 351 the EU region for filtering purposes. 352 353 Returns dict with keys: workspace_id, workspace_name, slug, organization_id, 354 dataplane_group_id, dataplane_name, created_at, tombstone 355 Or None if workspace not found. 356 """ 357 # Resolve workspace ID alias (workspace_id is required, so resolved value is never None) 358 resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id) 359 assert resolved_workspace_id is not None # Type narrowing: workspace_id is required 360 361 return query_workspace_info(resolved_workspace_id) 362 363 364@mcp_tool( 365 read_only=True, 366 idempotent=True, 367) 368def query_prod_connector_versions( 369 connector_definition_id: Annotated[ 370 str, 371 Field(description="Connector definition UUID to list versions for"), 372 ], 373) -> list[dict[str, Any]]: 374 """List all versions for a connector definition. 375 376 Returns all published versions of a connector, ordered by last_published 377 date descending. Useful for understanding version history and finding 378 specific version IDs for pinning or rollout monitoring. 379 380 Returns list of dicts with keys: version_id, docker_image_tag, docker_repository, 381 release_stage, support_level, cdk_version, language, last_published, release_date 382 """ 383 return query_connector_versions(connector_definition_id) 384 385 386@mcp_tool( 387 read_only=True, 388 idempotent=True, 389) 390def query_prod_new_connector_releases( 391 days: Annotated[ 392 int, 393 Field(description="Number of days to look back (default: 7)", default=7), 394 ] = 7, 395 limit: Annotated[ 396 int, 397 Field(description="Maximum number of results (default: 100)", default=100), 398 ] = 100, 399) -> list[dict[str, Any]]: 400 """List recently published connector versions. 401 402 Returns connector versions published within the specified number of days. 403 Uses last_published timestamp which reflects when the version was actually 404 deployed to the registry (not the changelog date). 405 406 Returns list of dicts with keys: version_id, connector_definition_id, docker_repository, 407 docker_image_tag, last_published, release_date, release_stage, support_level, 408 cdk_version, language, created_at 409 """ 410 return query_new_connector_releases(days=days, limit=limit) 411 412 413@mcp_tool( 414 read_only=True, 415 idempotent=True, 416) 417def query_prod_actors_by_pinned_connector_version( 418 connector_version_id: Annotated[ 419 str, 420 Field(description="Connector version UUID to find pinned instances for"), 421 ], 422) -> list[dict[str, Any]]: 423 """List actors (sources/destinations) effectively pinned to a specific connector version. 424 425 Returns all actors that are effectively pinned to a specific connector version, 426 considering all scope levels: actor-level pins, workspace-level pins, and 427 organization-level pins (with actor > workspace > organization precedence). 428 Useful for monitoring rollouts and understanding which customers are affected. 429 430 The actor_id field is the actor ID (superset of source_id/destination_id). 431 432 Returns list of dicts with keys: actor_id, connector_definition_id, origin_type, 433 origin, description, created_at, expires_at, pin_scope_type, actor_name, 434 workspace_id, workspace_name, organization_id, dataplane_group_id, dataplane_name 435 436 pin_scope_type is 'actor', 'workspace', or 'organization' indicating which scope 437 level the effective pin came from. 438 """ 439 return query_actors_pinned_to_version(connector_version_id) 440 441 442@mcp_tool( 443 read_only=True, 444 idempotent=True, 445) 446def query_prod_recent_syncs_for_connector_version( 447 connector_version_id: Annotated[ 448 str | None, 449 Field( 450 description=( 451 "Connector version UUID. Provide this OR " 452 "connector_name + connector_version." 453 ), 454 default=None, 455 ), 456 ] = None, 457 connector_name: Annotated[ 458 str | None, 459 Field( 460 description=( 461 "Canonical connector name (e.g. `source-pokeapi`, " 462 "`destination-duckdb`). Used with `connector_version` to " 463 "resolve the version UUID." 464 ), 465 default=None, 466 ), 467 ] = None, 468 connector_version: Annotated[ 469 str | None, 470 Field( 471 description=( 472 "Semver version tag (e.g. `0.3.59`). " 473 "Used with `connector_name` to resolve the version UUID." 474 ), 475 default=None, 476 ), 477 ] = None, 478 days: Annotated[ 479 int, 480 Field(description="Number of days to look back (default: 7)", default=7), 481 ] = 7, 482 limit: Annotated[ 483 int, 484 Field(description="Maximum number of results (default: 100)", default=100), 485 ] = 100, 486 successful_only: Annotated[ 487 bool, 488 Field( 489 description="If `True`, only return successful syncs (default: `False`)", 490 default=False, 491 ), 492 ] = False, 493) -> list[dict[str, Any]]: 494 """List sync jobs that were run with a specific connector version. 495 496 Works for both source and destination connectors. Automatically detects 497 the connector type from the version metadata and uses the appropriate 498 query variant. 499 500 Accepts either `connector_version_id` (UUID) or `connector_name` + 501 `connector_version` (e.g. `source-pokeapi` + `0.3.59`). When using 502 name + version, the `docker_repository` is derived from the canonical 503 name (e.g. `source-pokeapi` → `airbyte/source-pokeapi`). 504 505 Filters on the version stamped into `jobs.config` at job-creation time, 506 not the current pin state. This avoids false positives (pre-pin syncs 507 counted as RC) and false negatives (post-unpin syncs missed). 508 509 Pin columns (`pin_origin_type`, `pin_origin`, `pin_scope_type`) are 510 still included as informational output but are not used for filtering. 511 512 Returns list of dicts with keys: `job_id`, `connection_id`, `job_status`, 513 `started_at`, `job_updated_at`, `connection_name`, `actor_id`, `actor_name`, 514 `actor_definition_id`, `source_definition_version_id`, 515 `destination_definition_version_id`, `pin_origin_type`, 516 `pin_origin`, `pin_scope_type`, `workspace_id`, `workspace_name`, 517 `organization_id`, `dataplane_group_id`, `dataplane_name`. 518 """ 519 # Resolve inputs to a version UUID and connector type. 520 if connector_version_id is not None: 521 version_info = resolve_version_info(connector_version_id) 522 docker_repository = version_info["docker_repository"] 523 elif connector_name is not None and connector_version is not None: 524 # Derive docker_repository from canonical name. 525 docker_repository = f"airbyte/{connector_name}" 526 version_info = resolve_version_id_by_tag( 527 docker_repository=docker_repository, 528 docker_image_tag=connector_version, 529 ) 530 connector_version_id = version_info["version_id"] 531 else: 532 raise PyAirbyteInputError( 533 message=( 534 "Provide either `connector_version_id` or both " 535 "`connector_name` and `connector_version`." 536 ), 537 ) 538 539 is_destination = not is_source_connector(docker_repository) 540 return query_syncs_for_connector_version( 541 connector_version_id, 542 is_destination=is_destination, 543 days=days, 544 limit=limit, 545 successful_only=successful_only, 546 ) 547 548 549@mcp_tool( 550 read_only=True, 551 idempotent=True, 552 open_world=True, 553) 554def query_prod_recent_syncs_for_connector( 555 source_definition_id: Annotated[ 556 str | None, 557 Field( 558 description=( 559 "Source connector definition ID (UUID) to search for. " 560 "Provide this OR source_canonical_name OR destination_definition_id " 561 "OR destination_canonical_name (exactly one required). " 562 "Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics." 563 ), 564 default=None, 565 ), 566 ], 567 source_canonical_name: Annotated[ 568 str | None, 569 Field( 570 description=( 571 "Canonical source connector name to search for. " 572 "Provide this OR source_definition_id OR destination_definition_id " 573 "OR destination_canonical_name (exactly one required). " 574 "Examples: 'source-youtube-analytics', 'YouTube Analytics'." 575 ), 576 default=None, 577 ), 578 ], 579 destination_definition_id: Annotated[ 580 str | None, 581 Field( 582 description=( 583 "Destination connector definition ID (UUID) to search for. " 584 "Provide this OR destination_canonical_name OR source_definition_id " 585 "OR source_canonical_name (exactly one required). " 586 "Example: '94bd199c-2ff0-4aa2-b98e-17f0acb72610' for DuckDB." 587 ), 588 default=None, 589 ), 590 ], 591 destination_canonical_name: Annotated[ 592 str | None, 593 Field( 594 description=( 595 "Canonical destination connector name to search for. " 596 "Provide this OR destination_definition_id OR source_definition_id " 597 "OR source_canonical_name (exactly one required). " 598 "Examples: 'destination-duckdb', 'DuckDB'." 599 ), 600 default=None, 601 ), 602 ], 603 status_filter: Annotated[ 604 StatusFilter, 605 Field( 606 description=( 607 "Filter by job status: 'all' (default), 'succeeded', or 'failed'. " 608 "Use 'succeeded' to find healthy connections with recent successful syncs. " 609 "Use 'failed' to find connections with recent failures." 610 ), 611 default=StatusFilter.ALL, 612 ), 613 ], 614 organization_id: Annotated[ 615 str | OrganizationAliasEnum | None, 616 Field( 617 description=( 618 "Optional organization ID (UUID) or alias to filter results. " 619 "If provided, only syncs from this organization will be returned. " 620 "Accepts '@airbyte-internal' as an alias for the Airbyte internal org." 621 ), 622 default=None, 623 ), 624 ], 625 lookback_days: Annotated[ 626 int, 627 Field(description="Number of days to look back (default: 7)", default=7), 628 ], 629 limit: Annotated[ 630 int, 631 Field(description="Maximum number of results (default: 100)", default=100), 632 ], 633 customer_tier_filter: Annotated[ 634 TierFilter, 635 Field( 636 description=( 637 "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. " 638 "Filters results to only include connections belonging to organizations " 639 "in the specified tier. Use 'ALL' to include all tiers." 640 ), 641 ), 642 ] = "TIER_2", 643 *, 644 exclude_pinned: Annotated[ 645 bool, 646 Field( 647 description=( 648 "If True, exclude syncs for actors that are already pinned to a " 649 "specific version (at any scope level: actor, workspace, or organization). " 650 "Useful for 'prove fix' workflows where you want to find unpinned " 651 "connections for live testing. Default: False (include all syncs)." 652 ), 653 default=False, 654 ), 655 ], 656 enabled_schedules_only: Annotated[ 657 bool, 658 Field( 659 description=( 660 "If True, only return syncs for connections that are both active " 661 "(not paused/inactive) and on an automated sync schedule " 662 "(not manual-trigger-only). Useful for canary workflows where " 663 "you need connections that will produce organic syncs during a " 664 "monitoring window. Default: False (include all connections)." 665 ), 666 default=False, 667 ), 668 ], 669) -> list[dict[str, Any]]: 670 """List recent sync jobs for ALL actors using a connector type. 671 672 This tool finds all actors with the given connector definition and returns their 673 recent sync jobs, regardless of whether they have explicit version pins. It filters 674 out deleted actors, deleted workspaces, and deprecated connections. 675 676 Results are always enriched with customer_tier and is_eu fields. 677 The customer_tier_filter parameter is required to ensure tier-aware querying. 678 679 Use this tool to: 680 - Find healthy connections with recent successful syncs (status_filter='succeeded') 681 - Investigate connector issues across all users (status_filter='failed') 682 - Get an overview of all recent sync activity (status_filter='all') 683 684 Set `exclude_pinned=True` to filter out syncs for actors that are already pinned to a 685 specific version. This is useful for 'prove fix' live connection testing workflows 686 where you want to find unpinned connections to test against. 687 688 Set `enabled_schedules_only=True` to restrict results to connections that are both 689 enabled (status='active') and on an automated schedule (not manual-trigger-only). 690 This is useful for canary prerelease workflows where you need connections that 691 will run organically during the monitoring window. 692 693 Supports both SOURCE and DESTINATION connectors. Provide exactly one of: 694 source_definition_id, source_canonical_name, destination_definition_id, 695 or destination_canonical_name. 696 697 Key fields in results: 698 - job_status: 'succeeded', 'failed', 'cancelled', etc. 699 - connection_id, connection_name: The connection that ran the sync 700 - actor_id, actor_name: The source or destination actor 701 - customer_tier: TIER_0, TIER_1, or TIER_2 702 - is_eu: Whether the workspace is in the EU region 703 - pin_origin_type, pin_origin, pinned_version_id: Version pin context (NULL if not pinned) 704 - pin_scope_type: 'actor', 'workspace', or 'organization' (NULL if not pinned) 705 """ 706 # Validate that exactly one connector parameter is provided 707 provided_params = [ 708 source_definition_id, 709 source_canonical_name, 710 destination_definition_id, 711 destination_canonical_name, 712 ] 713 num_provided = sum(p is not None for p in provided_params) 714 if num_provided != 1: 715 raise PyAirbyteInputError( 716 message=( 717 "Exactly one of source_definition_id, source_canonical_name, " 718 "destination_definition_id, or destination_canonical_name must be provided." 719 ), 720 ) 721 722 # Determine if this is a destination connector 723 is_destination = ( 724 destination_definition_id is not None or destination_canonical_name is not None 725 ) 726 727 # Resolve canonical name to definition ID if needed 728 resolved_definition_id: str 729 if source_canonical_name: 730 resolved_definition_id = resolve_canonical_name_to_definition_id( 731 canonical_name=source_canonical_name, 732 ) 733 elif destination_canonical_name: 734 resolved_definition_id = resolve_canonical_name_to_definition_id( 735 canonical_name=destination_canonical_name, 736 ) 737 elif source_definition_id: 738 resolved_definition_id = source_definition_id 739 else: 740 # We've validated exactly one param is provided, so this must be set 741 assert destination_definition_id is not None 742 resolved_definition_id = destination_definition_id 743 744 # Resolve organization ID alias 745 resolved_organization_id = OrganizationAliasEnum.resolve(organization_id) 746 747 rows = query_recent_syncs_for_connector( 748 connector_definition_id=resolved_definition_id, 749 is_destination=is_destination, 750 status_filter=status_filter, 751 organization_id=resolved_organization_id, 752 days=lookback_days, 753 limit=limit, 754 exclude_pinned=exclude_pinned, 755 enabled_schedules_only=enabled_schedules_only, 756 ) 757 758 enriched = enrich_rows_by_org(rows) 759 return filter_rows_by_tier(enriched, customer_tier_filter) 760 761 762@mcp_tool( 763 read_only=True, 764 idempotent=True, 765 open_world=True, 766) 767def query_prod_failed_sync_attempts_for_connector( 768 source_definition_id: Annotated[ 769 str | None, 770 Field( 771 description=( 772 "Source connector definition ID (UUID) to search for. " 773 "Exactly one of this or source_canonical_name is required. " 774 "Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics." 775 ), 776 default=None, 777 ), 778 ] = None, 779 source_canonical_name: Annotated[ 780 str | None, 781 Field( 782 description=( 783 "Canonical source connector name to search for. " 784 "Exactly one of this or source_definition_id is required. " 785 "Examples: 'source-youtube-analytics', 'YouTube Analytics'." 786 ), 787 default=None, 788 ), 789 ] = None, 790 organization_id: Annotated[ 791 str | OrganizationAliasEnum | None, 792 Field( 793 description=( 794 "Optional organization ID (UUID) or alias to filter results. " 795 "If provided, only failed attempts from this organization will be returned. " 796 "Accepts '@airbyte-internal' as an alias for the Airbyte internal org." 797 ), 798 default=None, 799 ), 800 ] = None, 801 lookback_days: Annotated[ 802 int, 803 Field(description="Number of days to look back (default: 7)", default=7), 804 ] = 7, 805 limit: Annotated[ 806 int, 807 Field(description="Maximum number of results (default: 100)", default=100), 808 ] = 100, 809 customer_tier_filter: Annotated[ 810 TierFilter, 811 Field( 812 description=( 813 "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. " 814 "Filters results to only include connections belonging to organizations " 815 "in the specified tier. Use 'ALL' to include all tiers." 816 ), 817 ), 818 ] = "TIER_2", 819) -> list[dict[str, Any]]: 820 """List failed sync attempts for ALL actors using a source connector type. 821 822 This tool finds all actors with the given connector definition and returns their 823 failed sync attempts, regardless of whether they have explicit version pins. 824 825 Results are always enriched with customer_tier and is_eu fields. 826 The customer_tier_filter parameter is required to ensure tier-aware querying. 827 828 This is useful for investigating connector issues across all users. Use this when 829 you want to find failures for a connector type regardless of which version users 830 are on. 831 832 Note: This tool only supports SOURCE connectors. For destination connectors, 833 a separate tool would be needed. 834 835 Key fields in results: 836 - failure_summary: JSON containing failure details including failureType and messages 837 - customer_tier: TIER_0, TIER_1, or TIER_2 838 - is_eu: Whether the workspace is in the EU region 839 - pin_origin_type, pin_origin, pinned_version_id: Version pin context (NULL if not pinned) 840 - pin_scope_type: 'actor', 'workspace', or 'organization' (NULL if not pinned) 841 """ 842 # Validate that exactly one of the two parameters is provided 843 if (source_definition_id is None) == (source_canonical_name is None): 844 raise PyAirbyteInputError( 845 message=( 846 "Exactly one of source_definition_id or source_canonical_name " 847 "must be provided, but not both." 848 ), 849 ) 850 851 # Resolve canonical name to definition ID if needed 852 resolved_definition_id: str 853 if source_canonical_name: 854 resolved_definition_id = resolve_canonical_name_to_definition_id( 855 canonical_name=source_canonical_name, 856 ) 857 else: 858 resolved_definition_id = source_definition_id # ty: ignore[invalid-assignment] 859 860 # Resolve organization ID alias 861 resolved_organization_id = OrganizationAliasEnum.resolve(organization_id) 862 863 rows = query_failed_sync_attempts_for_connector( 864 connector_definition_id=resolved_definition_id, 865 organization_id=resolved_organization_id, 866 days=lookback_days, 867 limit=limit, 868 ) 869 enriched = enrich_rows_by_org(rows) 870 return filter_rows_by_tier(enriched, customer_tier_filter) 871 872 873@mcp_tool( 874 read_only=True, 875 idempotent=True, 876 open_world=True, 877) 878def query_prod_connections_by_connector( 879 source_definition_id: Annotated[ 880 str | None, 881 Field( 882 description=( 883 "Source connector definition ID (UUID) to search for. " 884 "Exactly one of source_definition_id, source_canonical_name, " 885 "destination_definition_id, or destination_canonical_name is required. " 886 "Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics." 887 ), 888 default=None, 889 ), 890 ] = None, 891 source_canonical_name: Annotated[ 892 str | None, 893 Field( 894 description=( 895 "Canonical source connector name to search for. " 896 "Exactly one of source_definition_id, source_canonical_name, " 897 "destination_definition_id, or destination_canonical_name is required. " 898 "Examples: 'source-youtube-analytics', 'YouTube Analytics'." 899 ), 900 default=None, 901 ), 902 ] = None, 903 destination_definition_id: Annotated[ 904 str | None, 905 Field( 906 description=( 907 "Destination connector definition ID (UUID) to search for. " 908 "Exactly one of source_definition_id, source_canonical_name, " 909 "destination_definition_id, or destination_canonical_name is required. " 910 "Example: 'e5c8e66c-a480-4a5e-9c0e-e8e5e4c5c5c5' for DuckDB." 911 ), 912 default=None, 913 ), 914 ] = None, 915 destination_canonical_name: Annotated[ 916 str | None, 917 Field( 918 description=( 919 "Canonical destination connector name to search for. " 920 "Exactly one of source_definition_id, source_canonical_name, " 921 "destination_definition_id, or destination_canonical_name is required. " 922 "Examples: 'destination-duckdb', 'DuckDB'." 923 ), 924 default=None, 925 ), 926 ] = None, 927 organization_id: Annotated[ 928 str | OrganizationAliasEnum | None, 929 Field( 930 description=( 931 "Optional organization ID (UUID) or alias to filter results. " 932 "If provided, only connections in this organization will be returned. " 933 "Accepts '@airbyte-internal' as an alias for the Airbyte internal org." 934 ), 935 default=None, 936 ), 937 ] = None, 938 limit: Annotated[ 939 int, 940 Field(description="Maximum number of results (default: 1000)", default=1000), 941 ] = 1000, 942 customer_tier_filter: Annotated[ 943 TierFilter, 944 Field( 945 description=( 946 "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. " 947 "Filters results to only include connections belonging to organizations " 948 "in the specified tier. Use 'ALL' to include all tiers." 949 ), 950 ), 951 ] = "TIER_2", 952 *, 953 exclude_pinned: Annotated[ 954 bool, 955 Field( 956 description=( 957 "If True, exclude connections whose connector is already pinned to a " 958 "specific version (at any scope level: actor, workspace, or organization). " 959 "Useful for 'prove fix' workflows where you want to find unpinned " 960 "connections for live testing. Default: False (include all connections)." 961 ), 962 default=False, 963 ), 964 ], 965 enabled_schedules_only: Annotated[ 966 bool, 967 Field( 968 description=( 969 "If True, only return connections that are both active " 970 "(not paused/inactive) and on an automated sync schedule " 971 "(not manual-trigger-only). Useful for canary workflows where " 972 "you need connections that will produce organic syncs during a " 973 "monitoring window. Default: False (include all connections)." 974 ), 975 default=False, 976 ), 977 ], 978) -> list[dict[str, Any]]: 979 """Search for all connections using a specific source or destination connector type. 980 981 This tool queries the Airbyte Cloud Prod DB Replica directly for fast results. 982 It finds all connections where the source or destination connector matches the 983 specified type, regardless of how the connector is named by users. 984 985 Results are always enriched with customer_tier and is_eu fields. 986 The customer_tier_filter parameter is required to ensure tier-aware querying. 987 988 Optionally filter by organization_id to limit results to a specific organization. 989 Use '@airbyte-internal' as an alias for the Airbyte internal organization. 990 991 Set `exclude_pinned=True` to filter out connections that are already pinned to a 992 specific version. This is useful for 'prove fix' live connection testing workflows 993 where you want to find unpinned connections to test against. 994 995 Set `enabled_schedules_only=True` to restrict results to connections that are both 996 enabled (status='active') and on an automated schedule (not manual-trigger-only). 997 This is useful for canary prerelease workflows where you need connections that 998 will run organically during the monitoring window. 999 1000 Returns a list of connection dicts with workspace context and clickable Cloud UI URLs. 1001 For source queries, returns: connection_id, connection_name, connection_url, source_id, 1002 source_name, source_definition_id, workspace_id, workspace_name, organization_id, 1003 dataplane_group_id, dataplane_name, pin_origin_type, pin_origin, pinned_version_id, 1004 pin_scope_type, customer_tier, is_eu. 1005 For destination queries, returns: connection_id, connection_name, connection_url, 1006 destination_id, destination_name, destination_definition_id, workspace_id, 1007 workspace_name, organization_id, dataplane_group_id, dataplane_name, pin_origin_type, 1008 pin_origin, pinned_version_id, pin_scope_type, customer_tier, is_eu. 1009 1010 pin_scope_type is 'actor', 'workspace', or 'organization' indicating which scope 1011 level the effective pin came from (NULL if not pinned). 1012 """ 1013 # Validate that exactly one of the four connector parameters is provided 1014 provided_params = [ 1015 source_definition_id, 1016 source_canonical_name, 1017 destination_definition_id, 1018 destination_canonical_name, 1019 ] 1020 num_provided = sum(p is not None for p in provided_params) 1021 if num_provided != 1: 1022 raise PyAirbyteInputError( 1023 message=( 1024 "Exactly one of source_definition_id, source_canonical_name, " 1025 "destination_definition_id, or destination_canonical_name must be provided." 1026 ), 1027 ) 1028 1029 # Determine if this is a source or destination query and resolve the definition ID 1030 is_source_query = ( 1031 source_definition_id is not None or source_canonical_name is not None 1032 ) 1033 resolved_definition_id: str 1034 1035 if source_canonical_name: 1036 resolved_definition_id = resolve_canonical_name_to_definition_id( 1037 canonical_name=source_canonical_name, 1038 ) 1039 elif source_definition_id: 1040 resolved_definition_id = source_definition_id 1041 elif destination_canonical_name: 1042 resolved_definition_id = resolve_canonical_name_to_definition_id( 1043 canonical_name=destination_canonical_name, 1044 ) 1045 else: 1046 resolved_definition_id = destination_definition_id # ty: ignore[invalid-assignment] 1047 1048 # Resolve organization ID alias 1049 resolved_organization_id = OrganizationAliasEnum.resolve(organization_id) 1050 1051 # Query the database based on connector type 1052 if is_source_query: 1053 rows = [ 1054 { 1055 "organization_id": str(row.get("organization_id", "")), 1056 "workspace_id": str(row["workspace_id"]), 1057 "workspace_name": row.get("workspace_name", ""), 1058 "connection_id": str(row["connection_id"]), 1059 "connection_name": row.get("connection_name", ""), 1060 "connection_url": ( 1061 f"{CLOUD_UI_BASE_URL}/workspaces/{row['workspace_id']}" 1062 f"/connections/{row['connection_id']}/status" 1063 ), 1064 "source_id": str(row["source_id"]), 1065 "source_name": row.get("source_name", ""), 1066 "source_definition_id": str(row["source_definition_id"]), 1067 "dataplane_group_id": str(row.get("dataplane_group_id", "")), 1068 "dataplane_name": row.get("dataplane_name", ""), 1069 "pin_origin_type": row.get("pin_origin_type"), 1070 "pin_origin": row.get("pin_origin"), 1071 "pinned_version_id": _opt_str(row.get("pinned_version_id")), 1072 "pin_scope_type": row.get("pin_scope_type"), 1073 } 1074 for row in query_connections_by_connector( 1075 connector_definition_id=resolved_definition_id, 1076 organization_id=resolved_organization_id, 1077 limit=limit, 1078 exclude_pinned=exclude_pinned, 1079 enabled_schedules_only=enabled_schedules_only, 1080 ) 1081 ] 1082 else: 1083 # Destination query 1084 rows = [ 1085 { 1086 "organization_id": str(row.get("organization_id", "")), 1087 "workspace_id": str(row["workspace_id"]), 1088 "workspace_name": row.get("workspace_name", ""), 1089 "connection_id": str(row["connection_id"]), 1090 "connection_name": row.get("connection_name", ""), 1091 "connection_url": ( 1092 f"{CLOUD_UI_BASE_URL}/workspaces/{row['workspace_id']}" 1093 f"/connections/{row['connection_id']}/status" 1094 ), 1095 "destination_id": str(row["destination_id"]), 1096 "destination_name": row.get("destination_name", ""), 1097 "destination_definition_id": str(row["destination_definition_id"]), 1098 "dataplane_group_id": str(row.get("dataplane_group_id", "")), 1099 "dataplane_name": row.get("dataplane_name", ""), 1100 "pin_origin_type": row.get("pin_origin_type"), 1101 "pin_origin": row.get("pin_origin"), 1102 "pinned_version_id": _opt_str(row.get("pinned_version_id")), 1103 "pin_scope_type": row.get("pin_scope_type"), 1104 } 1105 for row in query_connections_by_destination_connector( 1106 connector_definition_id=resolved_definition_id, 1107 organization_id=resolved_organization_id, 1108 limit=limit, 1109 exclude_pinned=exclude_pinned, 1110 enabled_schedules_only=enabled_schedules_only, 1111 ) 1112 ] 1113 1114 enriched = enrich_rows_by_org(rows) 1115 return filter_rows_by_tier(enriched, customer_tier_filter) 1116 1117 1118@mcp_tool( 1119 read_only=True, 1120 idempotent=True, 1121 open_world=True, 1122) 1123def query_prod_connections_by_stream( 1124 stream_name: Annotated[ 1125 str, 1126 Field( 1127 description=( 1128 "Name of the stream to search for in connection catalogs. " 1129 "This must match the exact stream name as configured in the connection. " 1130 "Examples: 'global_exclusions', 'campaigns', 'users'." 1131 ), 1132 ), 1133 ], 1134 source_definition_id: Annotated[ 1135 str | None, 1136 Field( 1137 description=( 1138 "Source connector definition ID (UUID) to search for. " 1139 "Provide this OR source_canonical_name (exactly one required). " 1140 "Example: 'afa734e4-3571-11ec-991a-1e0031268139' for YouTube Analytics." 1141 ), 1142 default=None, 1143 ), 1144 ], 1145 source_canonical_name: Annotated[ 1146 str | None, 1147 Field( 1148 description=( 1149 "Canonical source connector name to search for. " 1150 "Provide this OR source_definition_id (exactly one required). " 1151 "Examples: 'source-klaviyo', 'Klaviyo', 'source-youtube-analytics'." 1152 ), 1153 default=None, 1154 ), 1155 ], 1156 organization_id: Annotated[ 1157 str | OrganizationAliasEnum | None, 1158 Field( 1159 description=( 1160 "Optional organization ID (UUID) or alias to filter results. " 1161 "If provided, only connections in this organization will be returned. " 1162 "Accepts '@airbyte-internal' as an alias for the Airbyte internal org." 1163 ), 1164 default=None, 1165 ), 1166 ], 1167 limit: Annotated[ 1168 int, 1169 Field(description="Maximum number of results (default: 100)", default=100), 1170 ], 1171 customer_tier_filter: Annotated[ 1172 TierFilter, 1173 Field( 1174 description=( 1175 "Required tier filter: 'TIER_0', 'TIER_1', 'TIER_2', or 'ALL'. " 1176 "Filters results to only include connections belonging to organizations " 1177 "in the specified tier. Use 'ALL' to include all tiers." 1178 ), 1179 ), 1180 ] = "TIER_2", 1181) -> list[dict[str, Any]]: 1182 """Find connections that have a specific stream enabled in their catalog. 1183 1184 This tool searches the connection's configured catalog (JSONB) for streams 1185 matching the specified name. It's particularly useful when validating 1186 connector fixes that affect specific streams - you can quickly find 1187 customer connections that use the affected stream. 1188 1189 Results are always enriched with customer_tier and is_eu fields. 1190 The customer_tier_filter parameter is required to ensure tier-aware querying. 1191 1192 Use cases: 1193 - Finding connections with a specific stream enabled for regression testing 1194 - Validating connector fixes that affect particular streams 1195 - Identifying which customers use rarely-enabled streams 1196 1197 Returns a list of connection dicts with workspace context and clickable Cloud UI URLs. 1198 """ 1199 provided_params = [source_definition_id, source_canonical_name] 1200 num_provided = sum(p is not None for p in provided_params) 1201 if num_provided != 1: 1202 raise PyAirbyteInputError( 1203 message=( 1204 "Exactly one of source_definition_id or source_canonical_name " 1205 "must be provided." 1206 ), 1207 ) 1208 1209 resolved_definition_id: str 1210 if source_canonical_name: 1211 resolved_definition_id = resolve_canonical_name_to_definition_id( 1212 canonical_name=source_canonical_name, 1213 ) 1214 else: 1215 assert source_definition_id is not None 1216 resolved_definition_id = source_definition_id 1217 1218 resolved_organization_id = OrganizationAliasEnum.resolve(organization_id) 1219 1220 rows = [ 1221 { 1222 "organization_id": str(row.get("organization_id", "")), 1223 "workspace_id": str(row["workspace_id"]), 1224 "workspace_name": row.get("workspace_name", ""), 1225 "connection_id": str(row["connection_id"]), 1226 "connection_name": row.get("connection_name", ""), 1227 "connection_status": row.get("connection_status", ""), 1228 "connection_url": ( 1229 f"{CLOUD_UI_BASE_URL}/workspaces/{row['workspace_id']}" 1230 f"/connections/{row['connection_id']}/status" 1231 ), 1232 "source_id": str(row["source_id"]), 1233 "source_name": row.get("source_name", ""), 1234 "source_definition_id": str(row["source_definition_id"]), 1235 "dataplane_group_id": str(row.get("dataplane_group_id", "")), 1236 "dataplane_name": row.get("dataplane_name", ""), 1237 } 1238 for row in query_connections_by_stream( 1239 connector_definition_id=resolved_definition_id, 1240 stream_name=stream_name, 1241 organization_id=resolved_organization_id, 1242 limit=limit, 1243 ) 1244 ] 1245 enriched = enrich_rows_by_org(rows) 1246 return filter_rows_by_tier(enriched, customer_tier_filter) 1247 1248 1249@mcp_tool( 1250 read_only=True, 1251 idempotent=True, 1252) 1253def query_prod_organizations( 1254 name_contains: Annotated[ 1255 str, 1256 Field( 1257 description=( 1258 "Case-insensitive substring to search for in organization name or email. " 1259 "For example, 'acme' will match organizations named 'Acme Corp' or " 1260 "with email 'admin@acme.io'." 1261 ), 1262 ), 1263 ], 1264 limit: Annotated[ 1265 int, 1266 Field( 1267 description="Maximum number of organizations to return (default: 20)", 1268 default=20, 1269 ), 1270 ] = 20, 1271) -> OrganizationSearchResult: 1272 """Search organizations by name or email substring. 1273 1274 Performs a case-insensitive substring match on organization name and email. 1275 Use the returned `organization_id` values with other tools like 1276 `query_prod_connections_by_connector` or `lookup_customer_tiers`. 1277 """ 1278 rows = search_organizations(name_contains=name_contains, limit=limit) 1279 1280 orgs = [ 1281 OrganizationSearchHit( 1282 organization_id=str(row["organization_id"]), 1283 organization_name=row.get("organization_name", ""), 1284 email=row.get("email"), 1285 ) 1286 for row in rows 1287 ] 1288 1289 # Enrich with tier annotation 1290 org_ids = [o.organization_id for o in orgs] 1291 tier_results = {r.organization_id: r for r in get_org_tiers(org_ids)} 1292 for org in orgs: 1293 tier_result = tier_results.get(org.organization_id) 1294 if tier_result: 1295 org.customer_tier = tier_result.customer_tier 1296 1297 return OrganizationSearchResult( 1298 name_contains=name_contains, 1299 total_found=len(orgs), 1300 organizations=orgs, 1301 ) 1302 1303 1304@mcp_tool( 1305 read_only=True, 1306 idempotent=True, 1307) 1308def query_prod_workspaces( 1309 name_contains: Annotated[ 1310 str | None, 1311 Field( 1312 description=( 1313 "Case-insensitive substring to search for in workspace name or slug. " 1314 "For example, 'acme' will match workspaces named 'Acme Staging'." 1315 ), 1316 default=None, 1317 ), 1318 ] = None, 1319 email_domain: Annotated[ 1320 str | None, 1321 Field( 1322 description=( 1323 "Email domain to search for (e.g., 'motherduck.com'). " 1324 "Do not include the '@' symbol." 1325 ), 1326 default=None, 1327 ), 1328 ] = None, 1329 limit: Annotated[ 1330 int, 1331 Field( 1332 description="Maximum number of workspaces to return (default: 100)", 1333 default=100, 1334 ), 1335 ] = 100, 1336) -> WorkspaceSearchResult: 1337 """Search workspaces by name substring or email domain. 1338 1339 At least one of `name_contains` or `email_domain` must be provided. 1340 When `name_contains` is given, performs a case-insensitive substring match 1341 on workspace name and slug. When `email_domain` is given, matches 1342 workspaces by user email domain. 1343 1344 The returned organization IDs can be used with other tools like 1345 `query_prod_connections_by_connector` to find connections within 1346 those organizations for safe testing. 1347 """ 1348 if not name_contains and not email_domain: 1349 raise PyAirbyteInputError( 1350 message="At least one of `name_contains` or `email_domain` must be provided.", 1351 ) 1352 1353 if name_contains: 1354 rows = search_workspaces(name_contains=name_contains, limit=limit) 1355 else: 1356 assert email_domain is not None 1357 clean_domain = email_domain.lstrip("@") 1358 rows = query_workspaces_by_email_domain(email_domain=clean_domain, limit=limit) 1359 1360 workspaces = [ 1361 WorkspaceInfo( 1362 organization_id=str(row["organization_id"]), 1363 workspace_id=str(row["workspace_id"]), 1364 workspace_name=row.get("workspace_name", ""), 1365 slug=row.get("slug"), 1366 email=row.get("email"), 1367 dataplane_group_id=_opt_str(row.get("dataplane_group_id")), 1368 dataplane_name=row.get("dataplane_name"), 1369 created_at=row.get("created_at"), 1370 ) 1371 for row in rows 1372 ] 1373 1374 # Enrich with tier annotation (annotation only, no filtering) 1375 unique_org_ids = list(dict.fromkeys(w.organization_id for w in workspaces)) 1376 tier_results = {r.organization_id: r for r in get_org_tiers(unique_org_ids)} 1377 for ws in workspaces: 1378 tier_result = tier_results.get(ws.organization_id) 1379 if tier_result: 1380 ws.customer_tier = tier_result.customer_tier 1381 ws.is_eu = ws.dataplane_name == "EU" if ws.dataplane_name else False 1382 1383 return WorkspaceSearchResult( 1384 name_contains=name_contains, 1385 email_domain=email_domain.lstrip("@") if email_domain else None, 1386 total_workspaces_found=len(workspaces), 1387 unique_organization_ids=unique_org_ids, 1388 workspaces=workspaces, 1389 ) 1390 1391 1392# Backward-compatible alias 1393query_prod_workspaces_by_email_domain = query_prod_workspaces 1394 1395 1396def _build_connector_stats( 1397 connector_definition_id: str, 1398 connector_type: str, 1399 canonical_name: str | None, 1400 rows: list[dict[str, Any]], 1401 version_tags: dict[str, str | None], 1402) -> ConnectorConnectionStats: 1403 """Build ConnectorConnectionStats from query result rows.""" 1404 # Aggregate totals across all version groups 1405 total_connections = 0 1406 enabled_connections = 0 1407 active_connections = 0 1408 pinned_connections = 0 1409 unpinned_connections = 0 1410 total_succeeded = 0 1411 total_failed = 0 1412 total_cancelled = 0 1413 total_running = 0 1414 total_unknown = 0 1415 1416 by_version: list[VersionPinStats] = [] 1417 1418 for row in rows: 1419 version_id = row.get("pinned_version_id") 1420 row_total = int(row.get("total_connections", 0)) 1421 row_enabled = int(row.get("enabled_connections", 0)) 1422 row_active = int(row.get("active_connections", 0)) 1423 row_pinned = int(row.get("pinned_connections", 0)) 1424 row_unpinned = int(row.get("unpinned_connections", 0)) 1425 row_succeeded = int(row.get("succeeded_connections", 0)) 1426 row_failed = int(row.get("failed_connections", 0)) 1427 row_cancelled = int(row.get("cancelled_connections", 0)) 1428 row_running = int(row.get("running_connections", 0)) 1429 row_unknown = int(row.get("unknown_connections", 0)) 1430 1431 total_connections += row_total 1432 enabled_connections += row_enabled 1433 active_connections += row_active 1434 pinned_connections += row_pinned 1435 unpinned_connections += row_unpinned 1436 total_succeeded += row_succeeded 1437 total_failed += row_failed 1438 total_cancelled += row_cancelled 1439 total_running += row_running 1440 total_unknown += row_unknown 1441 1442 by_version.append( 1443 VersionPinStats( 1444 pinned_version_id=str(version_id) if version_id else None, 1445 docker_image_tag=version_tags.get(str(version_id)) 1446 if version_id 1447 else None, 1448 total_connections=row_total, 1449 enabled_connections=row_enabled, 1450 active_connections=row_active, 1451 latest_attempt=LatestAttemptBreakdown( 1452 succeeded=row_succeeded, 1453 failed=row_failed, 1454 cancelled=row_cancelled, 1455 running=row_running, 1456 unknown=row_unknown, 1457 ), 1458 ) 1459 ) 1460 1461 return ConnectorConnectionStats( 1462 connector_definition_id=connector_definition_id, 1463 connector_type=connector_type, 1464 canonical_name=canonical_name, 1465 total_connections=total_connections, 1466 enabled_connections=enabled_connections, 1467 active_connections=active_connections, 1468 pinned_connections=pinned_connections, 1469 unpinned_connections=unpinned_connections, 1470 latest_attempt=LatestAttemptBreakdown( 1471 succeeded=total_succeeded, 1472 failed=total_failed, 1473 cancelled=total_cancelled, 1474 running=total_running, 1475 unknown=total_unknown, 1476 ), 1477 by_version=by_version, 1478 ) 1479 1480 1481@mcp_tool( 1482 read_only=True, 1483 idempotent=True, 1484 open_world=True, 1485) 1486def query_prod_connector_connection_stats( 1487 source_definition_ids: Annotated[ 1488 list[str] | None, 1489 Field( 1490 description=( 1491 "List of source connector definition IDs (UUIDs) to get stats for. " 1492 "Example: ['afa734e4-3571-11ec-991a-1e0031268139']" 1493 ), 1494 default=None, 1495 ), 1496 ] = None, 1497 destination_definition_ids: Annotated[ 1498 list[str] | None, 1499 Field( 1500 description=( 1501 "List of destination connector definition IDs (UUIDs) to get stats for. " 1502 "Example: ['94bd199c-2ff0-4aa2-b98e-17f0acb72610']" 1503 ), 1504 default=None, 1505 ), 1506 ] = None, 1507 lookback_days: Annotated[ 1508 int, 1509 Field( 1510 description=( 1511 "Number of days to look back for 'active' connections (default: 7). " 1512 "Connections with sync activity within this window are counted as active." 1513 ), 1514 default=7, 1515 ), 1516 ] = 7, 1517) -> ConnectorConnectionStatsResponse: 1518 """Get aggregate connection stats for multiple connectors. 1519 1520 Returns counts of connections grouped by pinned version for each connector, 1521 including: 1522 - Total, enabled, and active connection counts 1523 - Pinned vs unpinned breakdown 1524 - Latest attempt status breakdown (succeeded, failed, cancelled, running, unknown) 1525 1526 This tool is designed for release monitoring workflows. It allows you to: 1527 1. Query recently released connectors to identify which ones to monitor 1528 2. Get aggregate stats showing how many connections are using each version 1529 3. See health metrics (pass/fail) broken down by version 1530 1531 The `lookback_days` parameter controls the lookback window for: 1532 - Counting 'active' connections (those with recent sync activity) 1533 - Determining 'latest attempt status' (most recent attempt within the window) 1534 1535 Connections with no sync activity in the lookback window will have 1536 'unknown' status in the latest_attempt breakdown. 1537 """ 1538 # Initialize empty lists if None 1539 source_ids = source_definition_ids or [] 1540 destination_ids = destination_definition_ids or [] 1541 1542 if not source_ids and not destination_ids: 1543 raise PyAirbyteInputError( 1544 message=( 1545 "At least one of source_definition_ids or destination_definition_ids " 1546 "must be provided." 1547 ), 1548 ) 1549 1550 sources: list[ConnectorConnectionStats] = [] 1551 destinations: list[ConnectorConnectionStats] = [] 1552 1553 # Process source connectors 1554 for source_def_id in source_ids: 1555 # Get version info for tag lookup 1556 versions = query_connector_versions(source_def_id) 1557 version_tags = { 1558 str(v["version_id"]): v.get("docker_image_tag") for v in versions 1559 } 1560 1561 # Get aggregate stats 1562 rows = query_source_connection_stats(source_def_id, days=lookback_days) 1563 1564 sources.append( 1565 _build_connector_stats( 1566 connector_definition_id=source_def_id, 1567 connector_type="source", 1568 canonical_name=None, 1569 rows=rows, 1570 version_tags=version_tags, 1571 ) 1572 ) 1573 1574 # Process destination connectors 1575 for dest_def_id in destination_ids: 1576 # Get version info for tag lookup 1577 versions = query_connector_versions(dest_def_id) 1578 version_tags = { 1579 str(v["version_id"]): v.get("docker_image_tag") for v in versions 1580 } 1581 1582 # Get aggregate stats 1583 rows = query_destination_connection_stats(dest_def_id, days=lookback_days) 1584 1585 destinations.append( 1586 _build_connector_stats( 1587 connector_definition_id=dest_def_id, 1588 connector_type="destination", 1589 canonical_name=None, 1590 rows=rows, 1591 version_tags=version_tags, 1592 ) 1593 ) 1594 1595 return ConnectorConnectionStatsResponse( 1596 sources=sources, 1597 destinations=destinations, 1598 lookback_days=lookback_days, 1599 generated_at=datetime.now(timezone.utc), 1600 ) 1601 1602 1603# ============================================================================= 1604# Connector Rollout Models and Tools 1605# ============================================================================= 1606 1607 1608class ConnectorRolloutInfo(BaseModel): 1609 """Information about a connector rollout.""" 1610 1611 rollout_id: str = Field(description="The rollout UUID") 1612 actor_definition_id: str = Field(description="The connector definition UUID") 1613 state: str = Field( 1614 description="Rollout state: initialized, workflow_started, in_progress, " 1615 "paused, finalizing, succeeded, errored, failed_rolled_back, canceled" 1616 ) 1617 initial_rollout_pct: int | None = Field( 1618 default=None, description="Initial rollout percentage" 1619 ) 1620 current_target_rollout_pct: int | None = Field( 1621 default=None, description="Current target rollout percentage" 1622 ) 1623 final_target_rollout_pct: int | None = Field( 1624 default=None, description="Final target rollout percentage" 1625 ) 1626 has_breaking_changes: bool = Field( 1627 description="Whether the RC has breaking changes" 1628 ) 1629 max_step_wait_time_mins: int | None = Field( 1630 default=None, description="Maximum wait time between rollout steps in minutes" 1631 ) 1632 rollout_strategy: str | None = Field( 1633 default=None, description="Rollout strategy: manual, automated, overridden" 1634 ) 1635 updated_by_user_id: str | None = Field( 1636 default=None, 1637 description="User ID recorded as last updating the rollout", 1638 ) 1639 updated_by_user_name: str | None = Field( 1640 default=None, 1641 description="Name recorded as last updating the rollout", 1642 ) 1643 updated_by_user_email: str | None = Field( 1644 default=None, 1645 description="Email recorded as last updating the rollout", 1646 ) 1647 workflow_run_id: str | None = Field( 1648 default=None, description="Temporal workflow run ID" 1649 ) 1650 error_msg: str | None = Field(default=None, description="Error message if errored") 1651 failed_reason: str | None = Field( 1652 default=None, description="Reason for failure if failed" 1653 ) 1654 paused_reason: str | None = Field( 1655 default=None, description="Reason for pause if paused" 1656 ) 1657 tag: str | None = Field(default=None, description="Optional tag for the rollout") 1658 created_at: datetime | None = Field( 1659 default=None, description="When the rollout was created" 1660 ) 1661 updated_at: datetime | None = Field( 1662 default=None, description="When the rollout was last updated" 1663 ) 1664 completed_at: datetime | None = Field( 1665 default=None, description="When the rollout completed (if terminal)" 1666 ) 1667 expires_at: datetime | None = Field( 1668 default=None, description="When the rollout expires" 1669 ) 1670 rc_docker_image_tag: str | None = Field( 1671 default=None, description="Docker image tag of the release candidate" 1672 ) 1673 rc_docker_repository: str | None = Field( 1674 default=None, description="Docker repository of the release candidate" 1675 ) 1676 initial_docker_image_tag: str | None = Field( 1677 default=None, description="Docker image tag of the initial version" 1678 ) 1679 initial_docker_repository: str | None = Field( 1680 default=None, description="Docker repository of the initial version" 1681 ) 1682 filters: dict[str, Any] | None = Field( 1683 default=None, 1684 description="Raw rollout filters JSON (e.g., {'tierFilter': {'tier': 'TIER_0'}})", 1685 ) 1686 customer_tier: str | None = Field( 1687 default=None, 1688 description="Customer tier targeted by this rollout (extracted from filters), " 1689 "e.g., 'TIER_0', 'TIER_1'. None if no tier filter is set.", 1690 ) 1691 1692 1693def _parse_rollout_filters(filters_raw: Any) -> dict[str, Any] | None: 1694 """Parse the rollout filters field from a database row. 1695 1696 The filters field may be a JSON string, a dict, or None. 1697 """ 1698 if filters_raw is None: 1699 return None 1700 if isinstance(filters_raw, dict): 1701 return filters_raw 1702 if isinstance(filters_raw, str): 1703 try: 1704 parsed = json.loads(filters_raw) 1705 if isinstance(parsed, dict): 1706 return parsed 1707 except (json.JSONDecodeError, TypeError): 1708 pass 1709 return None 1710 1711 1712def _extract_tier_from_filters(filters_raw: Any) -> str | None: 1713 """Extract customer tier from rollout filters JSON. 1714 1715 Supports two formats: 1716 - Legacy: `{"tierFilter": {"tier": "TIER_0"}}` 1717 - Current: `{"customerTierFilters": [{"name": "TIER", "value": ["TIER_1"], "operator": "IN"}]}` 1718 """ 1719 parsed = _parse_rollout_filters(filters_raw) 1720 if parsed is None: 1721 return None 1722 1723 # Current format: customerTierFilters list 1724 tier_filters = parsed.get("customerTierFilters") 1725 if isinstance(tier_filters, list): 1726 for entry in tier_filters: 1727 if isinstance(entry, dict) and entry.get("name") == "TIER": 1728 values = entry.get("value") 1729 if isinstance(values, list) and len(values) == 1: 1730 return str(values[0]) 1731 if isinstance(values, list) and len(values) > 1: 1732 return ", ".join(str(v) for v in values) 1733 1734 # Legacy format: tierFilter dict 1735 tier_filter = parsed.get("tierFilter") 1736 if isinstance(tier_filter, dict): 1737 tier = tier_filter.get("tier") 1738 if isinstance(tier, str): 1739 return tier 1740 1741 return None 1742 1743 1744def _row_to_connector_rollout_info(row: dict[str, Any]) -> ConnectorRolloutInfo: 1745 """Convert a database row to a ConnectorRolloutInfo model.""" 1746 return ConnectorRolloutInfo( 1747 rollout_id=str(row["rollout_id"]), 1748 actor_definition_id=str(row["actor_definition_id"]), 1749 state=row["state"], 1750 initial_rollout_pct=row.get("initial_rollout_pct"), 1751 current_target_rollout_pct=row.get("current_target_rollout_pct"), 1752 final_target_rollout_pct=row.get("final_target_rollout_pct"), 1753 has_breaking_changes=row["has_breaking_changes"], 1754 max_step_wait_time_mins=row.get("max_step_wait_time_mins"), 1755 rollout_strategy=row.get("rollout_strategy"), 1756 updated_by_user_id=str(row["updated_by_user_id"]) 1757 if row.get("updated_by_user_id") is not None 1758 else None, 1759 updated_by_user_name=row.get("updated_by_user_name"), 1760 updated_by_user_email=row.get("updated_by_user_email"), 1761 workflow_run_id=row.get("workflow_run_id"), 1762 error_msg=row.get("error_msg"), 1763 failed_reason=row.get("failed_reason"), 1764 paused_reason=row.get("paused_reason"), 1765 tag=row.get("tag"), 1766 created_at=row.get("created_at"), 1767 updated_at=row.get("updated_at"), 1768 completed_at=row.get("completed_at"), 1769 expires_at=row.get("expires_at"), 1770 rc_docker_image_tag=row.get("rc_docker_image_tag"), 1771 rc_docker_repository=row.get("rc_docker_repository"), 1772 initial_docker_image_tag=row.get("initial_docker_image_tag"), 1773 initial_docker_repository=row.get("initial_docker_repository"), 1774 filters=_parse_rollout_filters(row.get("filters")), 1775 customer_tier=_extract_tier_from_filters(row.get("filters")), 1776 ) 1777 1778 1779@mcp_tool( 1780 read_only=True, 1781 idempotent=True, 1782) 1783def query_prod_connector_rollouts( 1784 actor_definition_id: Annotated[ 1785 str | None, 1786 Field(description="Connector definition UUID to filter by (optional)"), 1787 ] = None, 1788 rollout_id: Annotated[ 1789 str | None, 1790 Field(description="Specific rollout UUID to look up (optional)"), 1791 ] = None, 1792 active_only: Annotated[ 1793 bool, 1794 Field(description="If true, only return active (non-terminal) rollouts"), 1795 ] = False, 1796 limit: Annotated[ 1797 int, 1798 Field(description="Maximum number of results (default: 100)"), 1799 ] = 100, 1800) -> list[ConnectorRolloutInfo]: 1801 """Query connector rollouts with flexible filtering. 1802 1803 Returns rollouts based on the provided filters. If no filters are specified, 1804 returns all active rollouts. Useful for monitoring rollout status and history. 1805 1806 Filter behavior: 1807 - rollout_id: Returns that specific rollout (ignores other filters) 1808 - active_only: Returns only active (non-terminal) rollouts 1809 - actor_definition_id: Returns rollouts for that specific connector 1810 - No filters: Returns all active rollouts (same as active_only=True) 1811 """ 1812 rows = query_connector_rollouts( 1813 actor_definition_id=actor_definition_id, 1814 rollout_id=rollout_id, 1815 active_only=active_only, 1816 limit=limit, 1817 ) 1818 return [_row_to_connector_rollout_info(row) for row in rows] 1819 1820 1821@mcp_tool( 1822 read_only=True, 1823 idempotent=True, 1824 open_world=True, 1825) 1826def query_prod_connection_sync_activity( 1827 start_at: Annotated[ 1828 datetime, 1829 Field( 1830 description=( 1831 "Inclusive start timestamp for the sync activity window. " 1832 "Must be timezone-aware (ISO 8601 with offset or `Z`)." 1833 ), 1834 ), 1835 ], 1836 end_at: Annotated[ 1837 datetime, 1838 Field( 1839 description=( 1840 "Exclusive end timestamp for the sync activity window. " 1841 "Must be timezone-aware and strictly after `start_at`." 1842 ), 1843 ), 1844 ], 1845 organization_id: Annotated[ 1846 str | OrganizationAliasEnum | None, 1847 Field( 1848 description=( 1849 "Optional organization UUID or alias. At least one of " 1850 "`organization_id`, `workspace_id`, or `connection_ids` is " 1851 "required. Accepts `@airbyte-internal` as an alias for the " 1852 "Airbyte internal org." 1853 ), 1854 default=None, 1855 ), 1856 ] = None, 1857 workspace_id: Annotated[ 1858 str | WorkspaceAliasEnum | None, 1859 Field( 1860 description=( 1861 "Optional workspace UUID or alias. At least one of " 1862 "`organization_id`, `workspace_id`, or `connection_ids` is " 1863 "required. Accepts `@devin-ai-sandbox` as an alias for the " 1864 "Devin AI sandbox workspace." 1865 ), 1866 default=None, 1867 ), 1868 ] = None, 1869 connection_ids: Annotated[ 1870 list[str] | None, 1871 Field( 1872 description=( 1873 "Optional list of connection UUIDs. At least one of " 1874 "`organization_id`, `workspace_id`, or `connection_ids` is " 1875 "required." 1876 ), 1877 default=None, 1878 ), 1879 ] = None, 1880 status_filter: Annotated[ 1881 StatusFilter, 1882 Field( 1883 description=( 1884 "Filter by job status: `all` (default), `succeeded`, or " 1885 "`failed`. Applied to `jobs.status` in the Prod DB Replica." 1886 ), 1887 default=StatusFilter.ALL, 1888 ), 1889 ] = StatusFilter.ALL, 1890 limit: Annotated[ 1891 int, 1892 Field( 1893 description="Maximum number of attempt rows to return.", 1894 default=1000, 1895 ), 1896 ] = 1000, 1897) -> list[dict[str, Any]]: 1898 """List recent sync jobs and attempts from the Prod DB Replica. 1899 1900 Returns one row per `(job, attempt)` pair for sync jobs whose `updated_at` 1901 falls in `[start_at, end_at)`, scoped to the provided organization, 1902 workspace, or connection IDs. Designed for live operational lookups — 1903 e.g. "what happened on this connection in the last hour" — not for 1904 historical analysis. 1905 1906 Each row is enriched with `customer_tier` and `is_eu` for the owning 1907 organization. Tier filtering is intentionally not applied — this is a 1908 read-only observability query. 1909 1910 Input requirements: 1911 - At least one of `organization_id`, `workspace_id`, or `connection_ids` 1912 must be provided (any combination is accepted). 1913 - `start_at` and `end_at` must be timezone-aware and `start_at < end_at`. 1914 1915 Key fields in each row: 1916 - `job_id`, `attempt_id`, `attempt_number` 1917 - `job_status`, `attempt_status` 1918 - `job_started_at`, `job_updated_at`, `attempt_ended_at` 1919 - `failure_summary` (JSON; populated when an attempt failed) 1920 - `connection_id`, `connection_name`, `connection_status` 1921 - `source_actor_id`, `source_actor_name`, `source_actor_definition_id` 1922 - `destination_actor_id`, `destination_actor_name`, 1923 `destination_actor_definition_id` 1924 - `workspace_id`, `workspace_name`, `organization_id` 1925 - `dataplane_group_id`, `dataplane_name` 1926 - `customer_tier`, `is_eu` (added by tier enrichment) 1927 """ 1928 resolved_organization_id = OrganizationAliasEnum.resolve(organization_id) 1929 resolved_workspace_id = WorkspaceAliasEnum.resolve(workspace_id) 1930 _validate_sync_activity_scope( 1931 organization_id=resolved_organization_id, 1932 workspace_id=resolved_workspace_id, 1933 connection_ids=connection_ids, 1934 ) 1935 normalized_start_at, normalized_end_at = _validate_sync_activity_window( 1936 start_at=start_at, 1937 end_at=end_at, 1938 ) 1939 1940 rows = query_connection_sync_activity_from_prod( 1941 start_at=normalized_start_at, 1942 end_at=normalized_end_at, 1943 organization_id=resolved_organization_id, 1944 workspace_id=resolved_workspace_id, 1945 connection_ids=connection_ids, 1946 status_filter=status_filter.value, 1947 limit=limit, 1948 ) 1949 return enrich_rows_by_org(rows) 1950 1951 1952# ============================================================================= 1953# Pinned Connector Versions Models and Tools 1954# ============================================================================= 1955 1956 1957class PinnedConnectorVersionInfo(BaseModel): 1958 """A connector version that has at least one scoped configuration pin.""" 1959 1960 version_id: str = Field(description="The actor_definition_version UUID") 1961 connector_definition_id: str = Field(description="The connector definition UUID") 1962 connector_name: str = Field(description="Human-readable connector name") 1963 docker_repository: str = Field(description="Docker repository path") 1964 docker_image_tag: str = Field(description="Docker image tag for this version") 1965 last_published: str | None = Field( 1966 default=None, description="ISO timestamp when this version was last published" 1967 ) 1968 pin_count: int = Field( 1969 description="Total number of scoped_configuration rows pinning to this version" 1970 ) 1971 breaking_change_pins: int = Field( 1972 default=0, 1973 description="Number of actor-scoped pins created by breaking changes", 1974 ) 1975 rollout_pins: int = Field( 1976 default=0, 1977 description="Number of pins created by connector rollouts", 1978 ) 1979 actor_pins: int = Field( 1980 description="Number of actor-scoped pins (excludes breaking change and rollout pins)" 1981 ) 1982 workspace_pins: int = Field(description="Number of workspace-scoped pins") 1983 org_pins: int = Field(description="Number of organization-scoped pins") 1984 1985 1986@mcp_tool( 1987 read_only=True, 1988 idempotent=True, 1989 open_world=True, 1990) 1991def query_connector_pin_stats( 1992 connector_definition_id: Annotated[ 1993 str | None, 1994 Field( 1995 description="Connector definition UUID to filter by (optional). " 1996 "Mutually exclusive with `connector_canonical_name`." 1997 ), 1998 ] = None, 1999 connector_canonical_name: Annotated[ 2000 str | None, 2001 Field( 2002 description="Connector canonical name (e.g. `source-postgres`) to filter by. " 2003 "Resolved to a definition ID via the registry. " 2004 "Mutually exclusive with `connector_definition_id`." 2005 ), 2006 ] = None, 2007) -> list[PinnedConnectorVersionInfo]: 2008 """Query connector versions that have at least one scoped configuration pin. 2009 2010 Returns versions from the prod DB that are referenced by at least one 2011 `scoped_configuration` pin (`key = 'connector_version'`). Each version 2012 appears exactly once with per-scope pin breakdown (actor, workspace, org). 2013 2014 If neither filter is provided, returns the global superset across all connectors. 2015 """ 2016 if connector_definition_id and connector_canonical_name: 2017 raise PyAirbyteInputError( 2018 message=( 2019 "Provide at most one of `connector_definition_id` or " 2020 "`connector_canonical_name`, not both." 2021 ), 2022 ) 2023 2024 resolved_id: str | None = None 2025 if connector_canonical_name: 2026 resolved_id = resolve_canonical_name_to_definition_id( 2027 canonical_name=connector_canonical_name, 2028 ) 2029 elif connector_definition_id: 2030 resolved_id = connector_definition_id 2031 2032 rows = query_versions_with_pins(actor_definition_id=resolved_id) 2033 return [ 2034 PinnedConnectorVersionInfo( 2035 version_id=str(row["version_id"]), 2036 connector_definition_id=str(row["connector_definition_id"]), 2037 connector_name=row["connector_name"], 2038 docker_repository=row["docker_repository"], 2039 docker_image_tag=row["docker_image_tag"], 2040 last_published=( 2041 row["last_published"].isoformat() if row.get("last_published") else None 2042 ), 2043 pin_count=row["pin_count"], 2044 breaking_change_pins=row.get("breaking_change_pins", 0), 2045 rollout_pins=row.get("rollout_pins", 0), 2046 actor_pins=row.get("actor_pins", 0), 2047 workspace_pins=row.get("workspace_pins", 0), 2048 org_pins=row.get("org_pins", 0), 2049 ) 2050 for row in rows 2051 ] 2052 2053 2054# ============================================================================= 2055# Organization-scoped Pin Models and Tools 2056# ============================================================================= 2057 2058 2059class PinOriginFilter(StrEnum): 2060 """How a `scoped_configuration` pin was created, used to filter pins.""" 2061 2062 ALL = "all" 2063 MANUAL = "manual" 2064 CONNECTOR_ROLLOUT = "connector_rollout" 2065 BREAKING_CHANGE = "breaking_change" 2066 2067 2068# Rollout states considered non-terminal ("active"). Kept in sync with the set 2069# used by the rollout-monitoring SQL in `airbyte_ops_mcp.prod_db_access.sql`. 2070_ACTIVE_ROLLOUT_STATES = frozenset( 2071 { 2072 "initialized", 2073 "workflow_started", 2074 "in_progress", 2075 "paused", 2076 "finalizing", 2077 "errored", 2078 } 2079) 2080 2081 2082def _pin_category(origin_type: str | None) -> str: 2083 """Classify a pin as `rollout`, `breaking_change`, or `manual`.""" 2084 if origin_type == "connector_rollout": 2085 return "rollout" 2086 if origin_type == "breaking_change": 2087 return "breaking_change" 2088 return "manual" 2089 2090 2091# Maps a `PinOriginFilter` to the `_pin_category` value it keeps. `origin_filter` 2092# is applied here in Python rather than in SQL — the fetched pin list is small, 2093# so an in-SQL `:origin_filter` OR-chain would only add scan cost. 2094_ORIGIN_FILTER_CATEGORIES: dict[str, str] = { 2095 "manual": "manual", 2096 "connector_rollout": "rollout", 2097 "breaking_change": "breaking_change", 2098} 2099 2100 2101def _resolve_connector_filter_id( 2102 *, 2103 connector_definition_id: str | None, 2104 connector_canonical_name: str | None, 2105) -> str | None: 2106 """Resolve mutually-exclusive connector inputs to a definition id or `None`. 2107 2108 Blank or whitespace-only inputs are treated as absent (`None`) so an 2109 optional filter passed as `""` does not become a real, zero-matching SQL 2110 filter. 2111 """ 2112 connector_definition_id = (connector_definition_id or "").strip() or None 2113 connector_canonical_name = (connector_canonical_name or "").strip() or None 2114 if connector_definition_id and connector_canonical_name: 2115 raise PyAirbyteInputError( 2116 message=( 2117 "Provide at most one of `connector_definition_id` or " 2118 "`connector_canonical_name`, not both." 2119 ), 2120 ) 2121 if connector_canonical_name: 2122 return resolve_canonical_name_to_definition_id( 2123 canonical_name=connector_canonical_name, 2124 ) 2125 return connector_definition_id 2126 2127 2128def _require_organization_id(organization_id: str | OrganizationAliasEnum) -> str: 2129 """Resolve a required organization id to a canonical UUID string. 2130 2131 Accepts an organization UUID or an `OrganizationAliasEnum` alias, resolving 2132 aliases to their UUID. The result is validated as a UUID and returned in 2133 canonical (lowercased) form for consistent logging and comparison. (The 2134 org-pin SQL casts this to native `uuid`, so matching is case-insensitive 2135 either way.) Raises `PyAirbyteInputError` on blank or malformed input rather 2136 than passing an invalid value to SQL and returning a confusing empty result. 2137 """ 2138 resolved = OrganizationAliasEnum.resolve(organization_id) 2139 if not resolved or not resolved.strip(): 2140 raise PyAirbyteInputError( 2141 message="`organization_id` is required (a non-empty organization UUID or alias).", 2142 ) 2143 try: 2144 return str(uuid.UUID(resolved.strip())) 2145 except ValueError as exc: 2146 raise PyAirbyteInputError( 2147 message="`organization_id` is not a valid organization UUID or alias.", 2148 ) from exc 2149 2150 2151def _normalize_optional_version_id(pinned_version_id: str | None) -> str | None: 2152 """Coerce a blank `pinned_version_id` to `None` and validate UUID format. 2153 2154 A blank or whitespace-only value is treated as "no filter" (`None`) rather 2155 than being passed to the SQL `uuid` cast, which would raise 2156 `invalid input syntax for type uuid`. 2157 """ 2158 normalized = (pinned_version_id or "").strip() or None 2159 if normalized is not None: 2160 try: 2161 uuid.UUID(normalized) 2162 except ValueError as exc: 2163 raise PyAirbyteInputError( 2164 message="`pinned_version_id` is not a valid UUID.", 2165 ) from exc 2166 return normalized 2167 2168 2169class OrgVersionPinStats(BaseModel): 2170 """A connector version pinned somewhere under an organization, with counts.""" 2171 2172 version_id: str = Field(description="The actor_definition_version UUID") 2173 connector_definition_id: str = Field(description="The connector definition UUID") 2174 connector_name: str = Field(description="Human-readable connector name") 2175 docker_repository: str = Field(description="Docker repository path") 2176 docker_image_tag: str = Field(description="Docker image tag for this version") 2177 last_published: str | None = Field( 2178 default=None, description="ISO timestamp when this version was last published" 2179 ) 2180 pin_count: int = Field( 2181 description="Total pins under the org targeting this version (all scopes)" 2182 ) 2183 manual_pins: int = Field( 2184 default=0, 2185 description="Pins with no system origin (user-created manual pins), any scope", 2186 ) 2187 rollout_pins: int = Field( 2188 default=0, description="Pins created by connector rollouts" 2189 ) 2190 breaking_change_pins: int = Field( 2191 default=0, description="Pins created by breaking changes" 2192 ) 2193 actor_pins: int = Field( 2194 description="Manual actor-scoped pins (excludes rollout and breaking-change)" 2195 ) 2196 workspace_pins: int = Field(description="Workspace-scoped pins under the org") 2197 org_pins: int = Field(description="Organization-scoped pins") 2198 has_active_rollout: bool = Field( 2199 default=False, 2200 description=( 2201 "`True` if at least one rollout pin is backed by a non-terminal " 2202 "`connector_rollout`" 2203 ), 2204 ) 2205 2206 2207class OrgConnectorPin(BaseModel): 2208 """A single `scoped_configuration` pin discovered under an organization.""" 2209 2210 connector_definition_id: str = Field(description="The connector definition UUID") 2211 connector_name: str = Field(description="Human-readable connector name") 2212 docker_repository: str = Field(description="Docker repository path") 2213 pinned_version_id: str = Field( 2214 description="The pinned actor_definition_version UUID" 2215 ) 2216 pinned_version_tag: str = Field( 2217 description="Docker image tag of the pinned version" 2218 ) 2219 pin_scope_type: str = Field( 2220 description="Scope of the pin: `organization`, `workspace`, or `actor`" 2221 ) 2222 scope_id: str = Field(description="UUID of the scoped entity") 2223 scope_name: str | None = Field( 2224 default=None, description="Display name of the scoped entity, when resolvable" 2225 ) 2226 pin_category: str = Field( 2227 description="Derived pin type: `manual`, `rollout`, or `breaking_change`" 2228 ) 2229 set_by: str | None = Field( 2230 default=None, 2231 description="Email (or name) of the user who set a manual pin, when known", 2232 ) 2233 rollout_id: str | None = Field( 2234 default=None, description="Backing connector_rollout UUID for rollout pins" 2235 ) 2236 rollout_state: str | None = Field( 2237 default=None, description="State of the backing rollout, for rollout pins" 2238 ) 2239 is_active_rollout: bool = Field( 2240 default=False, 2241 description="`True` when `rollout_state` is a non-terminal (active) state", 2242 ) 2243 description: str | None = Field(default=None, description="Free-text pin reason") 2244 reference_url: str | None = Field( 2245 default=None, description="Reference URL attached to the pin, when present" 2246 ) 2247 created_at: str | None = Field( 2248 default=None, description="ISO timestamp when the pin was created" 2249 ) 2250 expires_at: str | None = Field( 2251 default=None, description="ISO timestamp when the pin expires, when set" 2252 ) 2253 2254 2255@mcp_tool( 2256 read_only=True, 2257 idempotent=True, 2258 open_world=True, 2259) 2260def query_prod_pin_stats_for_organization( 2261 organization_id: Annotated[ 2262 str | OrganizationAliasEnum, 2263 Field( 2264 description="Organization UUID (or `@airbyte-internal` alias) to scope pins to. " 2265 "Resolve organization names to an ID first via `search_organizations`." 2266 ), 2267 ], 2268 connector_definition_id: Annotated[ 2269 str | None, 2270 Field( 2271 description="Connector definition UUID to filter by (optional). " 2272 "Mutually exclusive with `connector_canonical_name`." 2273 ), 2274 ] = None, 2275 connector_canonical_name: Annotated[ 2276 str | None, 2277 Field( 2278 description="Connector canonical name (e.g. `source-postgres`) to filter by. " 2279 "Resolved to a definition ID via the registry. " 2280 "Mutually exclusive with `connector_definition_id`." 2281 ), 2282 ] = None, 2283 limit: Annotated[ 2284 int, 2285 Field(description="Maximum number of versions to return (default: 1000)."), 2286 ] = 1000, 2287) -> list[OrgVersionPinStats]: 2288 """Query connector versions pinned anywhere under an organization. 2289 2290 Returns one row per pinned version, aggregating every `connector_version` 2291 pin whose scope belongs to the organization — the org itself, one of its 2292 workspaces, or an actor within one of those workspaces (actor, workspace, 2293 and organization scopes). Each row carries the per-scope pin breakdown, the 2294 manual/rollout/breaking-change split, and a `has_active_rollout` flag. 2295 2296 This powers the first step of the Organization Pins view (pick an org, then 2297 see the versions pinned under it). Use `query_prod_pins_for_organization` 2298 for the individual pins behind a selected version. 2299 """ 2300 resolved_org_id = _require_organization_id(organization_id) 2301 resolved_connector_id = _resolve_connector_filter_id( 2302 connector_definition_id=connector_definition_id, 2303 connector_canonical_name=connector_canonical_name, 2304 ) 2305 rows = query_org_pin_stats( 2306 resolved_org_id, 2307 connector_definition_id=resolved_connector_id, 2308 limit=limit, 2309 ) 2310 return [ 2311 OrgVersionPinStats( 2312 version_id=str(row["version_id"]), 2313 connector_definition_id=str(row["connector_definition_id"]), 2314 connector_name=row["connector_name"], 2315 docker_repository=row["docker_repository"], 2316 docker_image_tag=row["docker_image_tag"], 2317 last_published=( 2318 row["last_published"].isoformat() if row.get("last_published") else None 2319 ), 2320 pin_count=row["pin_count"], 2321 manual_pins=row.get("manual_pins", 0), 2322 rollout_pins=row.get("rollout_pins", 0), 2323 breaking_change_pins=row.get("breaking_change_pins", 0), 2324 actor_pins=row.get("actor_pins", 0), 2325 workspace_pins=row.get("workspace_pins", 0), 2326 org_pins=row.get("org_pins", 0), 2327 has_active_rollout=bool(row.get("has_active_rollout", False)), 2328 ) 2329 for row in rows 2330 ] 2331 2332 2333@mcp_tool( 2334 read_only=True, 2335 idempotent=True, 2336 open_world=True, 2337) 2338def query_prod_pins_for_organization( 2339 organization_id: Annotated[ 2340 str | OrganizationAliasEnum, 2341 Field( 2342 description="Organization UUID (or `@airbyte-internal` alias) to scope pins to. " 2343 "Resolve organization names to an ID first via `search_organizations`." 2344 ), 2345 ], 2346 connector_definition_id: Annotated[ 2347 str | None, 2348 Field( 2349 description="Connector definition UUID to filter by (optional). " 2350 "Mutually exclusive with `connector_canonical_name`." 2351 ), 2352 ] = None, 2353 connector_canonical_name: Annotated[ 2354 str | None, 2355 Field( 2356 description="Connector canonical name (e.g. `source-postgres`) to filter by. " 2357 "Resolved to a definition ID via the registry. " 2358 "Mutually exclusive with `connector_definition_id`." 2359 ), 2360 ] = None, 2361 pinned_version_id: Annotated[ 2362 str | None, 2363 Field( 2364 description="Actor_definition_version UUID to return only pins targeting " 2365 "that version. This is the post-selection filter for the org pins tab." 2366 ), 2367 ] = None, 2368 origin_filter: Annotated[ 2369 PinOriginFilter, 2370 Field( 2371 description="Restrict by how the pin was created: `all` (default), " 2372 "`manual`, `connector_rollout`, or `breaking_change`." 2373 ), 2374 ] = PinOriginFilter.ALL, 2375 limit: Annotated[ 2376 int, 2377 Field(description="Maximum number of pins to return (default: 1000)."), 2378 ] = 1000, 2379) -> list[OrgConnectorPin]: 2380 """List the individual connector-version pins discovered under an organization. 2381 2382 Returns one row per `scoped_configuration` pin whose scope belongs to the 2383 organization (org/workspace/actor), resolving the pinned connector and 2384 version, the scope's display name, the manual author's email, and — for 2385 rollout-origin pins — the backing `connector_rollout` id and state. This 2386 directly answers whether each pin is manual or caused by an active rollout. 2387 2388 This powers the second step of the Organization Pins view: after picking a 2389 version from `query_prod_pin_stats_for_organization`, pass its 2390 `pinned_version_id` here to list the pins behind it. 2391 """ 2392 resolved_org_id = _require_organization_id(organization_id) 2393 resolved_connector_id = _resolve_connector_filter_id( 2394 connector_definition_id=connector_definition_id, 2395 connector_canonical_name=connector_canonical_name, 2396 ) 2397 rows = query_org_connector_pins( 2398 resolved_org_id, 2399 connector_definition_id=resolved_connector_id, 2400 pinned_version_id=_normalize_optional_version_id(pinned_version_id), 2401 limit=limit, 2402 ) 2403 kept_category = _ORIGIN_FILTER_CATEGORIES.get(origin_filter.value) 2404 pins: list[OrgConnectorPin] = [] 2405 for row in rows: 2406 origin_type = row.get("origin_type") 2407 if kept_category is not None and _pin_category(origin_type) != kept_category: 2408 continue 2409 rollout_state = row.get("rollout_state") 2410 set_by = row.get("pinned_by_user_email") or row.get("pinned_by_user_name") 2411 created_at = row.get("created_at") 2412 expires_at = row.get("expires_at") 2413 pins.append( 2414 OrgConnectorPin( 2415 connector_definition_id=str(row["connector_definition_id"]), 2416 connector_name=row["connector_name"], 2417 docker_repository=row["docker_repository"], 2418 pinned_version_id=str(row["pinned_version_id"]), 2419 pinned_version_tag=row["pinned_version_tag"], 2420 pin_scope_type=str(row["pin_scope_type"]), 2421 scope_id=str(row["scope_id"]), 2422 scope_name=row.get("scope_name"), 2423 pin_category=_pin_category(origin_type), 2424 set_by=str(set_by) if set_by else None, 2425 rollout_id=str(row["rollout_id"]) if row.get("rollout_id") else None, 2426 rollout_state=rollout_state, 2427 is_active_rollout=rollout_state in _ACTIVE_ROLLOUT_STATES, 2428 description=row.get("description"), 2429 reference_url=row.get("reference_url"), 2430 created_at=created_at.isoformat() if created_at else None, 2431 expires_at=expires_at.isoformat() if expires_at else None, 2432 ) 2433 ) 2434 return pins 2435 2436 2437# ============================================================================= 2438# Health and Population Summary Models and Tools 2439# ============================================================================= 2440 2441 2442def _resolve_connector_target( 2443 *, 2444 connector_version_id: str | None, 2445 connector_name: str | None, 2446 connector_version: str | None, 2447 connector_definition_id: str | None, 2448 connector_canonical_name: str | None, 2449) -> tuple[str | None, str, str, str | None]: 2450 """Resolve mixed connector inputs to a common target tuple. 2451 2452 Returns `(version_id, definition_id, docker_repository, docker_image_tag)`. 2453 `version_id` and `docker_image_tag` are `None` when only a definition-level 2454 identifier was supplied. 2455 """ 2456 if connector_version_id is not None: 2457 info = resolve_version_info(connector_version_id) 2458 return ( 2459 connector_version_id, 2460 str(info["actor_definition_id"]), 2461 info["docker_repository"], 2462 info.get("docker_image_tag"), 2463 ) 2464 if connector_name is not None and connector_version is not None: 2465 docker_repository = f"airbyte/{connector_name}" 2466 info = resolve_version_id_by_tag( 2467 docker_repository=docker_repository, 2468 docker_image_tag=connector_version, 2469 ) 2470 return ( 2471 str(info["version_id"]), 2472 str(info["actor_definition_id"]), 2473 docker_repository, 2474 connector_version, 2475 ) 2476 2477 definition_id: str | None = None 2478 if connector_definition_id is not None: 2479 definition_id = connector_definition_id 2480 elif connector_canonical_name is not None: 2481 definition_id = resolve_canonical_name_to_definition_id( 2482 canonical_name=connector_canonical_name, 2483 ) 2484 if definition_id is None: 2485 raise PyAirbyteInputError( 2486 message=( 2487 "Provide one of: `connector_version_id`, " 2488 "`connector_name` + `connector_version`, " 2489 "`connector_definition_id`, or `connector_canonical_name`." 2490 ), 2491 ) 2492 versions = query_connector_versions(definition_id) 2493 if not versions: 2494 raise PyAirbyteInputError( 2495 message=f"No connector versions found for definition: {definition_id}", 2496 ) 2497 return (None, definition_id, versions[0]["docker_repository"], None) 2498 2499 2500class ConnectorPopulationSummary(BaseModel): 2501 """Applied vs potential pinning audience for a connector, split by tier.""" 2502 2503 connector_definition_id: str = Field(description="The connector definition UUID") 2504 connector_version_id: str | None = Field( 2505 default=None, 2506 description="The version UUID, when a specific version was requested", 2507 ) 2508 docker_repository: str = Field(description="Docker repository path") 2509 docker_image_tag: str | None = Field( 2510 default=None, description="Docker image tag, when a version was requested" 2511 ) 2512 customer_tier_filter: str = Field( 2513 description="Tier filter applied to the counts (`TIER_0`/`TIER_1`/`TIER_2`/`ALL`)" 2514 ) 2515 active: TierSummary = Field( 2516 description=( 2517 "Potential audience: enabled actors of the definition (those with at " 2518 "least one active connection, `status = 'active'`), by tier" 2519 ) 2520 ) 2521 pinned_any: TierSummary = Field( 2522 description="Active actors already pinned to any version, by tier" 2523 ) 2524 eligible: TierSummary = Field( 2525 description="Active actors not pinned to any version (available to pin), by tier" 2526 ) 2527 pinned_to_version: TierSummary | None = Field( 2528 default=None, 2529 description=( 2530 "Applied audience: actors pinned to the requested version, by tier. " 2531 "`None` when no specific version was requested." 2532 ), 2533 ) 2534 2535 2536class ConnectorVersionHealthSummary(BaseModel): 2537 """Four-bucket health rollup for the actors running a connector version.""" 2538 2539 connector_version_id: str = Field(description="The connector version UUID") 2540 connector_definition_id: str = Field(description="The connector definition UUID") 2541 docker_repository: str = Field(description="Docker repository path") 2542 docker_image_tag: str | None = Field( 2543 default=None, description="Docker image tag for this version" 2544 ) 2545 days: int = Field(description="Lookback window in days") 2546 customer_tier_filter: str = Field( 2547 description="Tier filter applied to the counts (`TIER_0`/`TIER_1`/`TIER_2`/`ALL`)" 2548 ) 2549 healthy: int = Field(description="Actors with at least one successful sync") 2550 unhealthy: int = Field( 2551 description="Actors with failures and no successes in the window" 2552 ) 2553 awaiting: int = Field( 2554 description="Actors that ran but produced only non-terminal jobs (no result yet)" 2555 ) 2556 disabled: int = Field( 2557 description=( 2558 "Actors pinned to the version that produced no jobs in the window — " 2559 "the dormant/inactive audience" 2560 ) 2561 ) 2562 total_actors: int = Field(description="Total actors counted across all states") 2563 healthy_by_tier: TierSummary = Field(description="Healthy actors by tier") 2564 unhealthy_by_tier: TierSummary = Field(description="Unhealthy actors by tier") 2565 awaiting_by_tier: TierSummary = Field(description="Awaiting-results actors by tier") 2566 disabled_by_tier: TierSummary = Field(description="Disabled actors by tier") 2567 2568 2569@mcp_tool( 2570 read_only=True, 2571 idempotent=True, 2572 open_world=True, 2573) 2574def query_connector_population_summary( 2575 connector_version_id: Annotated[ 2576 str | None, 2577 Field( 2578 description=( 2579 "Connector version UUID. When provided, the applied audience " 2580 "(`pinned_to_version`) is included. Provide this OR " 2581 "`connector_name` + `connector_version` OR a definition-level " 2582 "identifier." 2583 ), 2584 ), 2585 ] = None, 2586 connector_name: Annotated[ 2587 str | None, 2588 Field( 2589 description=( 2590 "Canonical connector name (e.g. `source-postgres`). Used with " 2591 "`connector_version` to resolve the version UUID." 2592 ), 2593 ), 2594 ] = None, 2595 connector_version: Annotated[ 2596 str | None, 2597 Field( 2598 description=( 2599 "Semver version tag (e.g. `0.3.59`). Used with `connector_name`." 2600 ), 2601 ), 2602 ] = None, 2603 connector_definition_id: Annotated[ 2604 str | None, 2605 Field( 2606 description=( 2607 "Connector definition UUID for a definition-level summary " 2608 "(no `pinned_to_version` breakdown)." 2609 ), 2610 ), 2611 ] = None, 2612 connector_canonical_name: Annotated[ 2613 str | None, 2614 Field( 2615 description=( 2616 "Canonical connector name resolved to a definition ID via the " 2617 "registry, for a definition-level summary." 2618 ), 2619 ), 2620 ] = None, 2621 customer_tier_filter: Annotated[ 2622 TierFilter, 2623 Field( 2624 description=( 2625 "Which customer tiers to count. Defaults to `TIER_2`; pass " 2626 "`ALL` to include TIER_0/TIER_1 (revenue-critical) customers." 2627 ), 2628 ), 2629 ] = "TIER_2", 2630) -> ConnectorPopulationSummary: 2631 """Summarize the applied vs potential pinning audience for a connector, by tier. 2632 2633 Answers "how many actors are pinned and how many are eligible for pinning, 2634 split by tier" — the population view analogous to a rollout's audience. 2635 2636 - `active`: the potential audience — non-tombstoned actors of the 2637 definition that have at least one *active* connection 2638 (`connection.status = 'active'`). Inactive/disabled and deprecated 2639 connections are excluded, so this reflects the enabled, rollout-touchable 2640 population rather than every actor ever created. 2641 - `pinned_any`: active actors that already have an effective 2642 `connector_version` pin at any scope (actor/workspace/org). 2643 - `eligible`: `active` minus `pinned_any` — actors available to pin. 2644 - `pinned_to_version`: the applied audience for the requested version 2645 (only when a version identifier was provided). 2646 2647 Backed by `scoped_configuration` + actor/connection tables, so it is cheap 2648 to compute. Accepts a version identifier (preferred, adds 2649 `pinned_to_version`) or a definition-level identifier. 2650 """ 2651 version_id, definition_id, docker_repository, docker_image_tag = ( 2652 _resolve_connector_target( 2653 connector_version_id=connector_version_id, 2654 connector_name=connector_name, 2655 connector_version=connector_version, 2656 connector_definition_id=connector_definition_id, 2657 connector_canonical_name=connector_canonical_name, 2658 ) 2659 ) 2660 is_destination = not is_source_connector(docker_repository) 2661 2662 population_rows = query_actor_population_by_org( 2663 actor_definition_id=definition_id, 2664 is_destination=is_destination, 2665 ) 2666 pinned_version_rows = ( 2667 query_actors_pinned_to_version(version_id) if version_id is not None else None 2668 ) 2669 summary = summarize_population( 2670 population_rows, 2671 pinned_version_rows=pinned_version_rows, 2672 tier_filter=customer_tier_filter, 2673 # The population query above is run without a `target_version_id`, so the 2674 # version-aware summaries do not apply (the per-version audience comes 2675 # from `pinned_version_rows` instead). 2676 has_target_version=False, 2677 ) 2678 return ConnectorPopulationSummary( 2679 connector_definition_id=definition_id, 2680 connector_version_id=version_id, 2681 docker_repository=docker_repository, 2682 docker_image_tag=docker_image_tag, 2683 customer_tier_filter=customer_tier_filter, 2684 active=summary.active_by_tier, 2685 pinned_any=summary.pinned_any_by_tier, 2686 eligible=summary.eligible_by_tier, 2687 pinned_to_version=summary.pinned_to_version_by_tier, 2688 ) 2689 2690 2691@mcp_tool( 2692 read_only=True, 2693 idempotent=True, 2694 open_world=True, 2695) 2696def query_connector_version_health_summary( 2697 connector_version_id: Annotated[ 2698 str | None, 2699 Field( 2700 description=( 2701 "Connector version UUID. Provide this OR " 2702 "`connector_name` + `connector_version`." 2703 ), 2704 ), 2705 ] = None, 2706 connector_name: Annotated[ 2707 str | None, 2708 Field( 2709 description=( 2710 "Canonical connector name (e.g. `source-postgres`). Used with " 2711 "`connector_version` to resolve the version UUID." 2712 ), 2713 ), 2714 ] = None, 2715 connector_version: Annotated[ 2716 str | None, 2717 Field( 2718 description=( 2719 "Semver version tag (e.g. `0.3.59`). Used with `connector_name`." 2720 ), 2721 ), 2722 ] = None, 2723 days: Annotated[ 2724 int, 2725 Field( 2726 description="Number of days to look back (default: 7, max: 30)", 2727 ge=1, 2728 le=30, 2729 ), 2730 ] = 7, 2731 include_pinned_disabled: Annotated[ 2732 bool, 2733 Field( 2734 description=( 2735 "If `True` (default), actors pinned to the version that ran no " 2736 "jobs in the window are counted as `disabled`." 2737 ), 2738 ), 2739 ] = True, 2740 customer_tier_filter: Annotated[ 2741 TierFilter, 2742 Field( 2743 description=( 2744 "Which customer tiers to count. Defaults to `TIER_2`; pass " 2745 "`ALL` to include TIER_0/TIER_1 (revenue-critical) customers." 2746 ), 2747 ), 2748 ] = "TIER_2", 2749) -> ConnectorVersionHealthSummary: 2750 """Summarize actor health for a connector version into four buckets. 2751 2752 Answers "how many actors on a version are healthy / unhealthy / awaiting / 2753 disabled". Classification per actor over the lookback window: 2754 2755 - `healthy`: at least one successful sync (the same success signal the 2756 autopilot health gate uses). 2757 - `unhealthy`: failures and no successes. 2758 - `awaiting`: ran but produced only non-terminal jobs (no result yet). 2759 - `disabled`: (when `include_pinned_disabled`) pinned to the version with no 2760 jobs at all in the window — the dormant/inactive audience. 2761 2762 Built on the attempt/version primitive — the version stamped into 2763 `jobs.config` at job-creation time — not the current pin state, so it 2764 reflects actors that actually ran this version. This scans jobs over the 2765 window and is more expensive than the population summary; keep `days` 2766 bounded and query per-version. 2767 """ 2768 if connector_version_id is None and not ( 2769 connector_name is not None and connector_version is not None 2770 ): 2771 raise PyAirbyteInputError( 2772 message=( 2773 "Provide either `connector_version_id` or both " 2774 "`connector_name` and `connector_version`." 2775 ), 2776 ) 2777 version_id, definition_id, docker_repository, docker_image_tag = ( 2778 _resolve_connector_target( 2779 connector_version_id=connector_version_id, 2780 connector_name=connector_name, 2781 connector_version=connector_version, 2782 connector_definition_id=None, 2783 connector_canonical_name=None, 2784 ) 2785 ) 2786 assert version_id is not None 2787 is_destination = not is_source_connector(docker_repository) 2788 2789 health_rows = query_version_actor_health( 2790 version_id, 2791 is_destination=is_destination, 2792 days=days, 2793 ) 2794 pinned_actor_rows = ( 2795 query_actors_pinned_to_version(version_id) if include_pinned_disabled else None 2796 ) 2797 summary = summarize_version_health( 2798 health_rows, 2799 pinned_actor_rows=pinned_actor_rows, 2800 tier_filter=customer_tier_filter, 2801 ) 2802 return ConnectorVersionHealthSummary( 2803 connector_version_id=version_id, 2804 connector_definition_id=definition_id, 2805 docker_repository=docker_repository, 2806 docker_image_tag=docker_image_tag, 2807 days=days, 2808 customer_tier_filter=customer_tier_filter, 2809 healthy=summary.healthy, 2810 unhealthy=summary.unhealthy, 2811 awaiting=summary.awaiting, 2812 disabled=summary.disabled, 2813 total_actors=summary.total_actors, 2814 healthy_by_tier=summary.healthy_by_tier, 2815 unhealthy_by_tier=summary.unhealthy_by_tier, 2816 awaiting_by_tier=summary.awaiting_by_tier, 2817 disabled_by_tier=summary.disabled_by_tier, 2818 ) 2819 2820 2821def register_prod_db_ops_tools(app: FastMCP) -> None: 2822 """Register prod DB query tools with the FastMCP app.""" 2823 register_mcp_tools(app, mcp_module=__name__)