22def register_interactive_tools(app: FastMCP) -> None:
23    """Register UI-presenting tools."""
24    register_mcp_tools(app, mcp_module="interactive")

 81@mcp_tool(
 82    read_only=True,
 83    idempotent=True,
 84    open_world=True,
 85    annotations={
 86        INTERACTIVE_UI_ANNOTATION: True,
 87    },
 88    app=PrefabAppConfig(),
 89)
 90def show_connectors_list(
 91    support_level: Annotated[
 92        str,
 93        Field(
 94            description=(
 95                "Exact support level to match, such as `certified`, `community`, "
 96                "or `archived`. Empty string means no filter."
 97            ),
 98        ),
 99    ] = "",
100    certified: Annotated[  # noqa: FBT002 - FastMCP tool parameter.
101        bool,
102        Field(
103            description=(
104                "When `True`, return only certified connectors. "
105                "Shorthand for `support_level='certified'`."
106            ),
107        ),
108    ] = False,
109    min_support_level: Annotated[
110        str,
111        Field(
112            description=(
113                "Minimum support level threshold. Levels: `archived` < `community` "
114                "< `certified`. Empty string means no filter."
115            ),
116        ),
117    ] = "",
118    connector_type: Annotated[
119        str,
120        Field(
121            description=(
122                "Filter by connector type: `source` or `destination`. "
123                "Empty string means no filter."
124            ),
125        ),
126    ] = "",
127    search: Annotated[
128        str,
129        Field(
130            description=(
131                "Case-insensitive search across connector name, display name, "
132                "definition ID, Docker repository, subtype, and docs URL."
133            ),
134        ),
135    ] = "",
136    limit: Annotated[
137        int,
138        Field(
139            description="Maximum number of connectors to return. Use `0` for no limit.",
140            ge=0,
141        ),
142    ] = 0,
143) -> ToolResult:
144    """Show an interactive public connector catalog from the OSS registry."""
145    if limit < 0:
146        raise exc.PyAirbyteInputError(
147            message="Limit parameter must be non-negative.",
148            context={"limit": limit},
149        )
150
151    eff_support_level = SupportLevel.CERTIFIED if certified else None
152    if support_level:
153        if certified:
154            raise ValueError(
155                "Cannot specify both `certified` and `support_level`. "
156                "Use `certified=True` as shorthand or `support_level` for explicit control."
157            )
158        eff_support_level = SupportLevel.parse(support_level)
159    eff_min_support_level = SupportLevel.parse(min_support_level) if min_support_level else None
160    if eff_support_level and eff_min_support_level:
161        raise ValueError(
162            "Cannot specify both `certified` or `support_level` and `min_support_level`. "
163            "Use an exact match or a threshold."
164        )
165    eff_connector_type = ConnectorType.parse(connector_type) if connector_type else None
166    filters = PublicConnectorFilters(
167        certified=certified,
168        support_level=eff_support_level.value if eff_support_level else None,
169        min_support_level=(eff_min_support_level.value if eff_min_support_level else None),
170        connector_type=eff_connector_type.value if eff_connector_type else None,
171        search=search,
172        limit=limit or None,
173    )
174    connectors = _list_public_registry_connectors(
175        support_level=eff_support_level,
176        min_support_level=eff_min_support_level,
177        connector_type=eff_connector_type,
178        search=search,
179        limit=limit or None,
180    )
181    registry_url = _get_registry_url()
182    raw_value = PublicConnectorListResult(
183        registry_url=registry_url,
184        connector_count=len(connectors),
185        filters=filters,
186        connectors=connectors,
187    )
188    agent_value = PublicConnectorListResult(
189        registry_url=raw_value.registry_url,
190        connector_count=raw_value.connector_count,
191        filters=raw_value.filters,
192        connectors=connectors[:CONNECTOR_CATALOG_AGENT_PREVIEW_LIMIT],
193    ).model_dump(mode="json")
194    model_preview_count = len(agent_value["connectors"])
195    full_count_rendered_to_user = len(connectors)
196    agent_value["model_preview_count"] = model_preview_count
197    agent_value["model_preview_limit"] = CONNECTOR_CATALOG_AGENT_PREVIEW_LIMIT
198    agent_value["model_preview_truncated"] = full_count_rendered_to_user > model_preview_count
199    agent_value["model_preview_omitted_count"] = full_count_rendered_to_user - model_preview_count
200    agent_value["full_count_rendered_to_user"] = full_count_rendered_to_user
201    agent_value["render_note"] = (
202        f"The `connectors` array is capped to {CONNECTOR_CATALOG_AGENT_PREVIEW_LIMIT} "
203        "entries for model context. The interactive widget renders all "
204        f"{full_count_rendered_to_user} matching connectors."
205    )
206    return ToolResult(
207        content=_json_dumps(_jsonable(agent_value)),
208        structured_content=connector_catalog_app(
209            connectors=connectors,
210            filters=filters,
211            registry_url=registry_url,
212        ),
213        meta={"airbyte_mcp_raw_result": _jsonable(agent_value)},
214    )

 76@mcp_tool(
 77    read_only=True,
 78    idempotent=True,
 79    open_world=True,
 80    annotations={
 81        INTERACTIVE_UI_ANNOTATION: True,
 82    },
 83    app=PrefabAppConfig(),
 84    extra_help_text=CLOUD_AUTH_TIP_TEXT,
 85)
 86def show_connection_sync_history(  # noqa: PLR0914
 87    ctx: Context,
 88    connection_id: Annotated[
 89        str,
 90        Field(description="The ID of the Airbyte Cloud connection to show sync history for."),
 91    ],
 92    *,
 93    workspace_id: Annotated[
 94        str | None,
 95        Field(
 96            description=WORKSPACE_ID_TIP_TEXT,
 97            default=None,
 98        ),
 99    ] = None,
100    max_jobs: Annotated[
101        int,
102        Field(
103            description=(
104                "Maximum number of recent sync jobs to display. "
105                "Defaults to 30. Maximum allowed value is 100."
106            ),
107            default=30,
108            ge=1,
109            le=100,
110        ),
111    ] = 30,
112    agent_context: Annotated[
113        Literal["verbose", "summary", "min"],
114        Field(
115            description=(
116                "Controls how much context is returned to the agent in the text response. "
117                "'verbose': full job-level data for detailed follow-up analysis. "
118                "'summary': aggregates and key observations only. "
119                "'min': one-liner confirmation that the dashboard rendered."
120            ),
121            default="min",
122        ),
123    ] = "min",
124    suppress_ui: Annotated[
125        bool,
126        Field(
127            description=(
128                "If True, skip rendering the visual dashboard and return only the agent "
129                "text response. Use this for follow-up data retrieval without re-rendering "
130                "the UI that the user has already seen."
131            ),
132            default=False,
133        ),
134    ] = False,
135) -> ToolResult:
136    """Show interactive sync history dashboard for an Airbyte Cloud connection.
137
138    Renders a rich UI with metrics (success rate, total records, total bytes),
139    charts (success/fail by date, records over time, bytes over time), and
140    a detailed job history table.
141    """
142    workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
143    connection = workspace.get_connection(connection_id=connection_id)
144    resolved_name = connection.name or connection_id
145    job_history_url = connection.job_history_url
146    source = connection.source
147    destination = connection.destination
148
149    sync_results = connection.get_previous_sync_logs(
150        limit=max_jobs,
151        from_tail=True,
152    )
153
154    jobs_data: list[dict[str, object]] = []
155    for sr in sync_results:
156        status = str(sr.get_job_status())
157        jobs_data.append(
158            {
159                "job_id": sr.job_id,
160                "status": status,
161                "bytes_synced": sr.bytes_synced,
162                "records_synced": sr.records_synced,
163                "start_time": sr.start_time.isoformat(),
164                "start_time_dt": sr.start_time,
165                "job_url": sr.job_url,
166            }
167        )
168
169    total_jobs = len(jobs_data)
170    succeeded = sum(1 for j in jobs_data if "succeeded" in str(j["status"]).lower())
171    success_rate = (succeeded / total_jobs * 100) if total_jobs > 0 else 0.0
172    total_records = sum(int(j["records_synced"]) for j in jobs_data)  # type: ignore[arg-type]
173    total_bytes = sum(int(j["bytes_synced"]) for j in jobs_data)  # type: ignore[arg-type]
174
175    # Per-job chart data (oldest first) for continuous timeline
176    jobs_chronological = list(reversed(jobs_data))
177    multi_day = len({str(j["start_time_dt"])[:10] for j in jobs_chronological}) > 1  # type: ignore[index]
178    chart_data: list[dict[str, int | str]] = [
179        {
180            "time": _time_label(j["start_time_dt"], include_date=multi_day),  # type: ignore[arg-type]
181            "succeeded": 1 if "succeeded" in str(j["status"]).lower() else 0,
182            "failed": 0 if "succeeded" in str(j["status"]).lower() else 1,
183            "records": int(j["records_synced"]),  # type: ignore[arg-type]
184            "bytes": int(j["bytes_synced"]),  # type: ignore[arg-type]
185        }
186        for j in jobs_chronological
187    ]
188
189    agent_text = _build_agent_text(
190        agent_context=agent_context,
191        connection_id=connection_id,
192        connection_name=resolved_name,
193        total_jobs=total_jobs,
194        succeeded=succeeded,
195        success_rate=success_rate,
196        total_records=total_records,
197        total_bytes=total_bytes,
198        jobs_data=jobs_data,
199    )
200
201    if suppress_ui:
202        return ToolResult(content=agent_text)
203
204    return ToolResult(
205        content=agent_text,
206        structured_content=_build_sync_history_app(
207            connection_name=resolved_name,
208            job_history_url=job_history_url or "",
209            source_name=source.name or "Source",
210            source_url=source.connector_url or "",
211            destination_name=destination.name or "Destination",
212            destination_url=destination.connector_url or "",
213            jobs_data=jobs_data,
214            chart_data=chart_data,
215            succeeded=succeeded,
216            success_rate=success_rate,
217            total_records=total_records,
218            total_bytes=total_bytes,
219        ),
220    )

117@mcp_tool(
118    read_only=True,
119    idempotent=True,
120    open_world=True,
121    annotations={
122        INTERACTIVE_UI_ANNOTATION: True,
123    },
124    app=PrefabAppConfig(),
125    extra_help_text=CLOUD_AUTH_TIP_TEXT,
126)
127def show_workspace_sync_status(
128    ctx: Context,
129    *,
130    workspace_id: Annotated[
131        str | None,
132        Field(
133            description=WORKSPACE_ID_TIP_TEXT,
134            default=None,
135        ),
136    ] = None,
137    max_connections: Annotated[
138        int,
139        Field(
140            description=(
141                "Maximum number of workspace connections to inspect. "
142                "Defaults to 50. Maximum allowed value is 100."
143            ),
144            default=50,
145            ge=1,
146            le=100,
147        ),
148    ] = 50,
149    max_jobs_per_connection: Annotated[
150        int,
151        Field(
152            description=(
153                "Maximum number of recent jobs to inspect for each connection. "
154                "Defaults to 5. Maximum allowed value is 10."
155            ),
156            default=5,
157            ge=1,
158            le=10,
159        ),
160    ] = 5,
161    recent_hours: Annotated[
162        int,
163        Field(
164            description=("Window, in hours, used for the Recently Synced metric. Defaults to 24."),
165            default=_RECENT_HOURS_DEFAULT,
166            ge=1,
167            le=720,
168        ),
169    ] = _RECENT_HOURS_DEFAULT,
170    agent_context: Annotated[
171        Literal["verbose", "summary", "min"],
172        Field(
173            description=(
174                "Controls how much context is returned to the agent in the text response. "
175                "'verbose': capped connection-level data for follow-up analysis. "
176                "'summary': aggregates and key observations only. "
177                "'min': one-liner confirmation that the dashboard rendered."
178            ),
179            default="min",
180        ),
181    ] = "min",
182    suppress_ui: Annotated[
183        bool,
184        Field(
185            description=(
186                "If True, skip rendering the visual dashboard and return only the agent "
187                "text response. Use this for follow-up data retrieval without re-rendering "
188                "the UI that the user has already seen."
189            ),
190            default=False,
191        ),
192    ] = False,
193) -> ToolResult:
194    """Show an interactive sync status dashboard for an Airbyte Cloud workspace."""
195    workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
196    connections = workspace.list_connections(limit=max_connections)
197    now = datetime.now(tz=timezone.utc)
198    connection_statuses = [
199        _summarize_connection(
200            connection=connection,
201            max_jobs_per_connection=max_jobs_per_connection,
202        )
203        for connection in connections
204    ]
205    metric_summary = _build_workspace_metric_summary(
206        connection_statuses=connection_statuses,
207        recent_hours=recent_hours,
208        now=now,
209    )
210    agent_text = _build_agent_text(
211        agent_context=agent_context,
212        workspace_id=workspace.workspace_id,
213        connection_statuses=connection_statuses,
214        metric_summary=metric_summary,
215    )
216    raw_result = {
217        **metric_summary,
218        "connections": [
219            asdict(connection_status)
220            for connection_status in connection_statuses[:WORKSPACE_SYNC_STATUS_AGENT_PREVIEW_LIMIT]
221        ],
222        "model_preview_count": min(
223            len(connection_statuses),
224            WORKSPACE_SYNC_STATUS_AGENT_PREVIEW_LIMIT,
225        ),
226        "model_preview_truncated": (
227            len(connection_statuses) > WORKSPACE_SYNC_STATUS_AGENT_PREVIEW_LIMIT
228        ),
229        "full_count_rendered_to_user": len(connection_statuses),
230        "suppress_ui": suppress_ui,
231    }
232
233    if suppress_ui:
234        return ToolResult(
235            content=agent_text,
236            meta={"airbyte_mcp_raw_result": raw_result},
237        )
238
239    return ToolResult(
240        content=agent_text,
241        structured_content=_build_workspace_sync_status_app(
242            workspace_id=workspace.workspace_id,
243            workspace_url=workspace.workspace_url or "",
244            connection_statuses=connection_statuses,
245            metric_summary=metric_summary,
246            recent_hours=recent_hours,
247        ),
248        meta={"airbyte_mcp_raw_result": raw_result},
249    )