diff --git a/airbyte/_direct_connectors/connector_docs.py b/airbyte/_direct_connectors/connector_docs.py
index 33eb4b60e..591a53845 100644
--- a/airbyte/_direct_connectors/connector_docs.py
+++ b/airbyte/_direct_connectors/connector_docs.py
@@ -6,7 +6,7 @@
 - `CloudWorkspace.get_agent_skill_docs` / `CloudConnector.read_agent_skill_docs`
   (MCP tool `get_external_access_skill_docs`) read destination skill docs straight from
   the Agents API, with no local enrichment.
-- `CloudConnector.get_direct_access_guidance` (MCP tool `describe_cloud_connector`
+- `CloudConnector.get_direct_access_guidance` (MCP tool `describe_connector`
   with `with_direct_access_guidance=True`) prepends a short local intro --
   location, table naming, and dialect notes -- to the server-served destination
   docs overview.
@@ -17,7 +17,7 @@
 
 This module also builds `_DirectConnectorInspectResult` payloads locally from the
 Cloud workspace objects and summarizes the connections touching a connector for
-the `describe_cloud_connector` MCP tool.
+the `describe_connector` MCP tool.
 """
 
 from __future__ import annotations
diff --git a/airbyte/_direct_connectors/models.py b/airbyte/_direct_connectors/models.py
index aa18348d1..8cc0bda9a 100644
--- a/airbyte/_direct_connectors/models.py
+++ b/airbyte/_direct_connectors/models.py
@@ -406,7 +406,7 @@ def _collect_index_errors(self) -> ExternalSearchResult:
         if not self.metadata and not self.errors:
             collected.append(
                 "No index matched this search type and stream selection, so empty results "
-                "do not mean there are no matches. Call `get_cloud_search_status` to see "
+                "do not mean there are no matches. Call `get_search_index_status` to see "
                 "which streams have which index types."
             )
         self.warnings = [
diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py
index 910896771..1da770873 100644
--- a/airbyte/_util/api_util.py
+++ b/airbyte/_util/api_util.py
@@ -72,7 +72,7 @@
     "This is expected right after creating a connector with `defer_credentials=True`. "
     "Don't retry yet. Ask the user to open the connector's settings page, complete "
     "authentication and any missing settings, then test and save. After they confirm, call "
-    "`check_cloud_connector`; once it succeeds, retry this call."
+    "`check_connector`; once it succeeds, retry this call."
 )
 ACTOR_NOT_READY_PYTHON_GUIDANCE = (
     "This is expected right after creating a connector as a draft. Don't retry yet. "
diff --git a/airbyte/_util/destination_smoke_tests.py b/airbyte/_util/destination_smoke_tests.py
index fb4bc8914..a0ee7bea4 100644
--- a/airbyte/_util/destination_smoke_tests.py
+++ b/airbyte/_util/destination_smoke_tests.py
@@ -3,7 +3,7 @@
 
 This module provides the core logic for running smoke tests against destination
 connectors. It is used by both the CLI (`pyab destination-smoke-test`) and the
-MCP tool (`destination_smoke_test`).
+MCP tool (`run_local_destination_smoke_test`).
 
 Smoke tests send synthetic data from the built-in smoke test source to a
 destination connector and report whether the destination accepted the data
diff --git a/airbyte/cloud/client.py b/airbyte/cloud/client.py
index 0c15a563d..3cabdff0f 100644
--- a/airbyte/cloud/client.py
+++ b/airbyte/cloud/client.py
@@ -246,7 +246,7 @@ def get_workspace(self, workspace_id: str | None = None) -> CloudWorkspace:
                 guidance=(
                     "No workspace was configured, and no default workspace could be resolved "
                     "for the authenticated user. Provide a workspace ID, or call "
-                    "`get_default_cloud_context` to discover your workspaces and organizations."
+                    "`get_default_workspace_context` to discover your workspaces and organizations."
                 ),
             )
 
@@ -907,13 +907,13 @@ def get_default_context_for_user(self) -> CloudDefaultContextInfo:
         if any(permission.get("permissionType") == "instance_admin" for permission in permissions):
             discovery_hints.append(
                 "Instance-admin access may include every organization and workspace in the "
-                "instance. Use list_cloud_organizations(name_contains=...) or "
-                "list_cloud_workspaces(organization_id=...) to discover others."
+                "instance. Use list_organizations(name_contains=...) or "
+                "list_workspaces(organization_id=...) to discover others."
             )
         if membership_organization_ids:
             discovery_hints.append(
                 "Organization membership grants access to every workspace in those "
-                "organizations. Use list_cloud_workspaces(organization_id=<id>) to "
+                "organizations. Use list_workspaces(organization_id=<id>) to "
                 "discover workspaces."
             )
         if user is not None and not isinstance(user.get("defaultWorkspaceId"), str):
@@ -956,13 +956,13 @@ def _missing_default_workspace_hint(self) -> str:
                 "Your Airbyte user profile has no default workspace; this "
                 "session uses the configured workspace "
                 f"{self.default_workspace_id}. Use "
-                "set_default_cloud_workspace(user_email=<your email>, "
+                "set_default_workspace(user_email=<your email>, "
                 "workspace_id=<id>) to set a durable default; it applies to "
                 "both MCP sessions and the Airbyte Cloud web app."
             )
         return (
             "No default workspace is set. Use "
-            "set_default_cloud_workspace(user_email=<your email>, "
+            "set_default_workspace(user_email=<your email>, "
             "workspace_id=<id>) to durably set one; it applies to both MCP "
             "sessions and the Airbyte Cloud web app."
         )
@@ -1008,7 +1008,7 @@ def set_default_workspace_for_user(
             raise exc.AirbyteLibInputError(
                 message=("The provided `user_email` does not match the authenticated user."),
                 guidance=(
-                    "Call get_default_cloud_context to see the authenticated user "
+                    "Call get_default_workspace_context to see the authenticated user "
                     "context, then pass that user's email."
                 ),
                 context={
@@ -1030,21 +1030,21 @@ def set_default_workspace_for_user(
         except exc.AirbyteMissingResourceError as error:
             raise exc.AirbyteLibInputError(
                 message=f"Workspace {workspace_id} was not found.",
-                guidance=("Call get_default_cloud_context to see your member workspaces."),
+                guidance=("Call get_default_workspace_context to see your member workspaces."),
                 context={"workspace_id": workspace_id},
             ) from error
         except exc.AirbyteCloudError as error:
             if (error.context or {}).get("status_code") == HTTPStatus.NOT_FOUND:
                 raise exc.AirbyteLibInputError(
                     message=f"Workspace {workspace_id} was not found.",
-                    guidance=("Call get_default_cloud_context to see your member workspaces."),
+                    guidance=("Call get_default_workspace_context to see your member workspaces."),
                     context={"workspace_id": workspace_id},
                 ) from error
             raise
         if workspace.get("tombstone"):
             raise exc.AirbyteLibInputError(
                 message=f"Workspace {workspace_id} is tombstoned (deleted).",
-                guidance=("Call get_default_cloud_context to see your member workspaces."),
+                guidance=("Call get_default_workspace_context to see your member workspaces."),
                 context={"workspace_id": workspace_id},
             )
         workspace_organization_id = workspace.get("organizationId")
@@ -1063,7 +1063,7 @@ def set_default_workspace_for_user(
                     f"or its organization {workspace_organization_id}."
                 ),
                 guidance=(
-                    "Call get_default_cloud_context to see your member_workspaces "
+                    "Call get_default_workspace_context to see your member_workspaces "
                     "and member_organizations, then choose a workspace you are a "
                     "member of."
                 ),
@@ -1173,7 +1173,7 @@ def _raise_ambiguous_organization_error(
                 "with one of these "
                 "organization IDs "
                 f"(showing {len(candidates)} of {len(organization_ids)}): {candidate_details}. "
-                "Call `get_default_cloud_context` to see your memberships."
+                "Call `get_default_workspace_context` to see your memberships."
             ),
             context={
                 "organization_ids": list(organization_ids),
@@ -1350,7 +1350,7 @@ def get_organization(
             raise exc.AirbyteLibInputError(
                 message="Organization ID or organization name is required.",
                 guidance=(
-                    "Provide an organization ID or name, or call `get_default_cloud_context` "
+                    "Provide an organization ID or name, or call `get_default_workspace_context` "
                     "to discover your organizations."
                 ),
             )
diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py
index 15d29f401..a67d764b4 100644
--- a/airbyte/cloud/connections.py
+++ b/airbyte/cloud/connections.py
@@ -371,7 +371,7 @@ def run_sync(
                     guidance=(
                         "Re-enable the connection first (e.g. "
                         "`connection.set_enabled(enabled=True)`, or the "
-                        "`update_cloud_connection` MCP tool with `enabled=True`), then retry."
+                        "`update_connection` MCP tool with `enabled=True`), then retry."
                     ),
                     context={"connection_id": self.connection_id},
                 ) from ex
@@ -599,7 +599,7 @@ def get_sync_result(
                         "or you don't have access to it."
                     ),
                     guidance=(
-                        "Use `list_cloud_sync_jobs` to find valid job IDs for this connection."
+                        "Use `list_connection_sync_jobs` to find valid job IDs for this connection."
                     ),
                     context={"connection_id": self.connection_id, "job_id": job_id},
                 ) from ex
@@ -612,7 +612,9 @@ def get_sync_result(
                 message=(
                     f"Job {job_id} belongs to a different connection, not {self.connection_id}."
                 ),
-                guidance="Use `list_cloud_sync_jobs` to find valid job IDs for this connection.",
+                guidance=(
+                    "Use `list_connection_sync_jobs` to find valid job IDs for this connection."
+                ),
                 context={"connection_id": self.connection_id, "job_id": job_id},
             )
 
diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py
index 0e88b5c57..1e45fc682 100644
--- a/airbyte/cloud/workspaces.py
+++ b/airbyte/cloud/workspaces.py
@@ -214,7 +214,7 @@ def __init__(
             raise exc.AirbyteLibInputError(
                 message="Workspace ID is required.",
                 guidance=(
-                    "Provide a workspace ID, or call `get_default_cloud_context` to discover "
+                    "Provide a workspace ID, or call `get_default_workspace_context` to discover "
                     "available workspaces."
                 ),
             )
diff --git a/airbyte/exceptions.py b/airbyte/exceptions.py
index 94f413cd6..db5762446 100644
--- a/airbyte/exceptions.py
+++ b/airbyte/exceptions.py
@@ -308,12 +308,12 @@ def __post_init__(self) -> None:
         if is_mcp_mode():
             self.guidance = (
                 "The authenticated user's default workspace was checked and none was "
-                "available. `list_cloud_workspaces` returns direct workspace memberships "
+                "available. `list_workspaces` returns direct workspace memberships "
                 "by default; pass `organization_id`/`organization_name` or a broader "
                 "`privilege_scope` for organization-wide discovery, or call "
-                "`list_cloud_organizations` to search organizations by name. If exactly "
+                "`list_organizations` to search organizations by name. If exactly "
                 "one workspace is found, use it; otherwise ask the user to choose. Call "
-                "`get_default_cloud_context` to inspect your memberships."
+                "`get_default_workspace_context` to inspect your memberships."
             )
         else:
             self.guidance = (
diff --git a/airbyte/mcp/_docs_results.py b/airbyte/mcp/_docs_results.py
index 06b8fabc3..6c30c9b58 100644
--- a/airbyte/mcp/_docs_results.py
+++ b/airbyte/mcp/_docs_results.py
@@ -24,7 +24,7 @@
 class CloudConnectorDocsResult(BaseModel):
     """Connector docs rendered for agent consumption by the Cloud MCP tools.
 
-    Returned by `describe_cloud_connector`.
+    Returned by `describe_connector`.
     """
 
     model_config = ConfigDict(extra="allow")
diff --git a/airbyte/mcp/_user_identity.py b/airbyte/mcp/_user_identity.py
index a6c13040e..e36a0018c 100644
--- a/airbyte/mcp/_user_identity.py
+++ b/airbyte/mcp/_user_identity.py
@@ -74,7 +74,7 @@
 class AirbyteUser:
     """Canonical Airbyte user identity and its default workspace, when available.
 
-    Cached without expiry (evicted when `set_default_cloud_workspace` changes it), so
+    Cached without expiry (evicted when `set_default_workspace` changes it), so
     `default_workspace_id` may be stale. It is only used to derive the organization
     recorded in telemetry; tools resolve the default workspace live.
     """
diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py
index 28787615e..759852b0d 100644
--- a/airbyte/mcp/cloud.py
+++ b/airbyte/mcp/cloud.py
@@ -163,7 +163,7 @@ def _manifest_and_testing_values_hint() -> str | None:
     "authentication methods, include the method's selector field in `config`. The result "
     "includes a settings link for the person to complete any missing fields and test the draft. "
     "After they report a successful test and save, call "
-    "`check_cloud_connector` with the `connector_id` (optionally `connector_type` and "
+    "`check_connector` with the `connector_id` (optionally `connector_type` and "
     "`workspace_id`)."
 )
 DEFERRED_SETUP_GUIDANCE = (
@@ -171,7 +171,7 @@ def _manifest_and_testing_values_hint() -> str | None:
     "missing settings, then test and save. Until then the connector is a draft: you can "
     "describe or delete it, but creating a connection, discovering streams, or running a "
     "sync fails with `409 actor-not-ready`. Don't attempt or retry those until the user "
-    "confirms. Then call `check_cloud_connector` with `connector_id` (optionally "
+    "confirms. Then call `check_connector` with `connector_id` (optionally "
     "`connector_type` and `workspace_id`); a successful check makes the connector ready "
     "to use."
 )
@@ -355,7 +355,7 @@ class CloudOrganizationResult(BaseModel):
     email: str | None = None
     """Email associated with the organization, when available.
 
-    Only returned by `describe_cloud_organization`; `list_cloud_organizations`
+    Only returned by `describe_organization`; `list_organizations`
     reports `email_domain` instead.
     """
     email_domain: str | None = None
@@ -567,7 +567,7 @@ class DeferredDeployResult(BaseModel):
     name: str
     """The connector name in Airbyte Cloud."""
     workspace_id: str
-    """The workspace the connector was created in; pass it to `check_cloud_connector`."""
+    """The workspace the connector was created in; pass it to `check_connector`."""
     settings_url: str
     """Cloud settings page where a person completes the credentials."""
     guidance: str = DEFERRED_SETUP_GUIDANCE
@@ -635,7 +635,7 @@ def _get_cloud_client(
 
 
 @mcp_tool()
-def deploy_connector_to_cloud(  # noqa: PLR0913  # Mirrors the API surface.
+def deploy_connector(  # noqa: PLR0913  # Mirrors the API surface.
     ctx: Context,
     name: Annotated[
         str,
@@ -890,7 +890,7 @@ def _get_suggested_streams_for_source(
 
 
 @mcp_tool()
-def create_connection_on_cloud(
+def create_connection(
     ctx: Context,
     connection_name: Annotated[
         str,
@@ -939,7 +939,7 @@ def create_connection_on_cloud(
     """Create a connection between a deployed source and destination on Airbyte Cloud.
 
     New connections use a manual schedule by default. Automatic syncs require a later
-    schedule update with `update_cloud_connection`.
+    schedule update with `update_connection`.
 
     When `selected_streams` is not provided, the connection selects the source
     connector's suggested streams from the connector registry. If the connector has
@@ -994,7 +994,7 @@ def create_connection_on_cloud(
 @mcp_tool(
     pipeline_change=False,
 )
-def run_cloud_sync(
+def run_connection_sync(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -1044,7 +1044,7 @@ def run_cloud_sync(
 
 
 @mcp_tool()
-def deploy_noop_destination_to_cloud(
+def deploy_noop_destination(
     ctx: Context,
     destination_name: Annotated[
         str,
@@ -1082,7 +1082,7 @@ def deploy_noop_destination_to_cloud(
     read_only=True,
     idempotent=True,
 )
-def get_cloud_sync_status(
+def get_connection_sync_status(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -1095,7 +1095,7 @@ def get_cloud_sync_status(
         Field(
             description=(
                 "Optional job ID; it must belong to `connection_id`. Use "
-                "`list_cloud_sync_jobs` to find job IDs for this connection."
+                "`list_connection_sync_jobs` to find job IDs for this connection."
             ),
             default=None,
         ),
@@ -1120,7 +1120,7 @@ def get_cloud_sync_status(
     """Get the status of a sync job from Airbyte Cloud.
 
     `connection_id` is required, and any supplied `job_id` must belong to that connection. Use
-    `list_cloud_sync_jobs` to find job IDs for the connection.
+    `list_connection_sync_jobs` to find job IDs for the connection.
     """
     workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id)
     connection = workspace.get_connection(connection_id=connection_id)
@@ -1162,7 +1162,7 @@ def get_cloud_sync_status(
     read_only=True,
     idempotent=True,
 )
-def list_cloud_sync_jobs(
+def list_connection_sync_jobs(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -1257,7 +1257,7 @@ def list_cloud_sync_jobs(
     destructive=True,
     pipeline_change=False,
 )
-def cancel_cloud_sync(
+def cancel_connection_sync(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -1327,7 +1327,7 @@ class CloudConnectorResult(BaseModel):
     read_only=True,
     idempotent=True,
 )
-def list_cloud_connectors(
+def list_connectors(
     ctx: Context,
     *,
     workspace_id: Annotated[
@@ -1371,7 +1371,8 @@ def list_cloud_connectors(
 ) -> list[CloudConnectorResult]:
     """List deployed source and destination connectors in the Airbyte Cloud workspace.
 
-    Use `list_cloud_connections` to find the pipelines between these connectors, and
+    Use `list_connector_definitions` for available connector types in the Airbyte registry.
+    Use `list_connections` to find the pipelines between these connectors, and
     `feature_filter` to find connectors with external-access features.
     """
     if limit is not None and limit <= 0:
@@ -1453,7 +1454,7 @@ def list_cloud_connectors(
 class CloudConnectorDetailsResult(BaseModel):
     """A description of a deployed Cloud connector.
 
-    As returned by the `describe_cloud_connector` MCP tool.
+    As returned by the `describe_connector` MCP tool.
     """
 
     model_config = ConfigDict(extra="allow")
@@ -1509,7 +1510,7 @@ def _describe_cloud_connector(  # noqa: PLR0912  # Too many branches
     with_direct_access_guidance: bool,
     with_data_replication_docs: bool,
 ) -> CloudConnectorDetailsResult:
-    """Assemble the `describe_cloud_connector` MCP tool's result for a deployed connector."""
+    """Assemble the `describe_connector` MCP tool's result for a deployed connector."""
     connector_type = connector.connector_type
     warnings: list[str] = []
     try:
@@ -1588,7 +1589,7 @@ def _describe_cloud_connector(  # noqa: PLR0912  # Too many branches
     read_only=True,
     idempotent=True,
 )
-def describe_cloud_connector(
+def describe_connector(
     ctx: Context,
     connector_id: Annotated[
         str,
@@ -1798,7 +1799,7 @@ def execute_external_api_query(  # noqa: PLR0913  # Explicit args mirror the con
     field is available before relying on it in a pipeline, or answering a one-off
     question without waiting for a sync.
 
-    `describe_cloud_connector` with `with_direct_access_guidance=True` also returns the
+    `describe_connector` with `with_direct_access_guidance=True` also returns the
     connector's action docs.
     """
     connector = _get_cloud_workspace(ctx, workspace_id).get_connector(connector_id)
@@ -1902,7 +1903,7 @@ def _execute_external_api_action(  # noqa: PLR0913  # Explicit args mirror the c
     Creates, updates, or deletes data in the external system.
 
     Before calling, read the connector's action docs with `get_external_access_skill_docs`
-    (or `describe_cloud_connector` with `with_direct_access_guidance=True`) to
+    (or `describe_connector` with `with_direct_access_guidance=True`) to
     learn the entity types, actions, and required `api_args`; argument names
     differ per connector and are not guessable.
     """
@@ -2050,8 +2051,8 @@ def _search_disabled_errors(
         ),
         "Once search is enabled, the connector's synced data must be indexed before "
         "searches return results. New syncs are indexed as they run; data already in the "
-        "destination is indexed only after `request_cloud_search_index_backfill`. Call "
-        "`get_cloud_search_status` to check which streams are indexed and whether their "
+        "destination is indexed only after `request_search_index_backfill`. Call "
+        "`get_search_index_status` to check which streams are indexed and whether their "
         "backfill is complete.",
     ]
 
@@ -2171,14 +2172,14 @@ def execute_external_search_query(  # noqa: PLR0913  # Explicit args mirror the
     A source searches its own indexed data; a destination searches every indexed source
     synced to it. On a destination search, every `streams` entry needs the `source_id`
     of the source that stream belongs to. When results are empty, call
-    `get_cloud_search_status` to check which streams are indexed and whether their
+    `get_search_index_status` to check which streams are indexed and whether their
     backfill is complete.
 
     Keep responses small: use `streams[].fields` to project each hit's `entity_data`
     (an output projection, not a match filter) and `max_context_chars` to truncate
     each hit's `context`. `min_similarity` and `max_similarity_diff` apply only to
     `semantic` and `hybrid` searches. Pass `destination_id` for a source synced to
-    several enabled destinations; the IDs are listed by `get_cloud_search_status`.
+    several enabled destinations; the IDs are listed by `get_search_index_status`.
     Per-index failures are reported in `warnings`.
 
     A keyword-only hit leaves out of `entity_data` any field longer than 4096 characters
@@ -2214,7 +2215,7 @@ def execute_external_search_query(  # noqa: PLR0913  # Explicit args mirror the
     idempotent=True,
     external_access=True,
 )
-def get_cloud_search_status(
+def get_search_index_status(
     ctx: Context,
     *,
     connector_id: Annotated[
@@ -2260,7 +2261,7 @@ def get_cloud_search_status(
     search indexes. A stream is searchable once it has at least one index.
 
     Turning search on indexes only data from new syncs; data already in the destination
-    is indexed only by a backfill requested with `request_cloud_search_index_backfill`. A
+    is indexed only by a backfill requested with `request_search_index_backfill`. A
     stream's `backfill` is null when none was requested. Otherwise `backfill.status` is
     `starting`, `running`, `finalizing`, `complete`, `failed`, or `cancelled`, with
     `failed_at` or `cancelled_at` set when the run ended that way. A stream whose table is
@@ -2318,7 +2319,7 @@ def matches(stream: ExternalSearchStreamStatus) -> bool:
     pipeline_change=False,
     external_access=True,
 )
-def request_cloud_search_index_backfill(
+def request_search_index_backfill(
     ctx: Context,
     *,
     connection_id: Annotated[
@@ -2371,7 +2372,7 @@ def request_cloud_search_index_backfill(
     run. Without `streams`, the result has one entry with an empty `stream_name` that
     stands for every stream.
 
-    Track progress with `get_cloud_search_status` on the connection's source or
+    Track progress with `get_search_index_status` on the connection's source or
     destination: each stream's `backfill.status` moves from `starting` through `running`
     and `finalizing` to `complete`, `failed`, or `cancelled`. A stream whose table exists
     but is empty completes with 0 records. A run fails when, for example, the destination
@@ -2398,7 +2399,7 @@ def request_cloud_search_index_backfill(
     read_only=True,
     idempotent=True,
 )
-def check_cloud_connector(
+def check_connector(
     ctx: Context,
     connector_id: Annotated[
         str,
@@ -2437,7 +2438,7 @@ def check_cloud_connector(
     read_only=True,
     idempotent=True,
 )
-def describe_cloud_connection(
+def describe_connection(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -2476,7 +2477,7 @@ def describe_cloud_connection(
     read_only=True,
     idempotent=True,
 )
-def get_cloud_sync_logs(
+def get_connection_sync_logs(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -2555,7 +2556,7 @@ def get_cloud_sync_logs(
                 resource_type="sync job",
                 resource_name_or_id=connection_id,
                 message=f"Connection '{connection_id}' has no sync jobs yet.",
-                guidance="Run a sync first with `run_cloud_sync`, then request its logs.",
+                guidance="Run a sync first with `run_connection_sync`, then request its logs.",
             )
         raise AirbyteMissingResourceError(
             resource_type="sync job",
@@ -2630,7 +2631,7 @@ def get_cloud_sync_logs(
     read_only=True,
     idempotent=True,
 )
-def list_cloud_connections(
+def list_connections(
     ctx: Context,
     *,
     workspace_id: Annotated[
@@ -2674,7 +2675,7 @@ def list_cloud_connections(
 ) -> list[CloudConnectionResult]:
     """List all deployed connections in the Airbyte Cloud workspace.
 
-    Each connection links a source to a destination; use `list_cloud_connectors` to list those
+    Each connection links a source to a destination; use `list_connectors` to list those
     deployed resources.
 
     When with_connection_status is True, each connection result will include
@@ -2757,7 +2758,7 @@ def list_cloud_connections(
     read_only=True,
     idempotent=True,
 )
-def list_cloud_workspaces(
+def list_workspaces(
     ctx: Context,
     *,
     organization_id: Annotated[
@@ -2845,7 +2846,7 @@ def list_cloud_workspaces(
         "No workspaces were returned for these credentials. By default only direct "
         "workspace memberships are listed; pass `organization_id` or a broader "
         "`privilege_scope` to discover organization-wide workspaces, or call "
-        "`get_default_cloud_context` to inspect your memberships."
+        "`get_default_workspace_context` to inspect your memberships."
         if not results
         else None
     )
@@ -2878,7 +2879,7 @@ def list_cloud_workspaces(
     read_only=True,
     idempotent=True,
 )
-def get_default_cloud_context(ctx: Context) -> CloudDefaultContextResult:
+def get_default_workspace_context(ctx: Context) -> CloudDefaultContextResult:
     """Return the authenticated user's default Cloud context.
 
     This is the one-call orientation entry point: it resolves the default
@@ -2930,13 +2931,13 @@ def get_default_cloud_context(ctx: Context) -> CloudDefaultContextResult:
     if truncated_memberships:
         message += (
             f" Only the first {' and '.join(truncated_memberships)} are shown; use "
-            "list_cloud_organizations or list_cloud_workspaces to see the rest."
+            "list_organizations or list_workspaces to see the rest."
         )
     if context.unvalidated_workspace_count > 0:
         message += (
             f" {context.unvalidated_workspace_count} additional direct workspace grant(s) were "
             f"not validated because this call checks at most {_MAX_WORKSPACES_TO_VALIDATE}; use "
-            "list_cloud_workspaces to see them."
+            "list_workspaces to see them."
         )
     if resolved_default_workspace is not None:
         message = resolved_default_workspace + message
@@ -2973,7 +2974,7 @@ def get_default_cloud_context(ctx: Context) -> CloudDefaultContextResult:
     idempotent=True,
     destructive=True,
 )
-def set_default_cloud_workspace(
+def set_default_workspace(
     ctx: Context,
     user_email: Annotated[
         str,
@@ -2981,7 +2982,7 @@ def set_default_cloud_workspace(
             description=(
                 "Email of the authenticated Airbyte Cloud user this change applies to. "
                 "Must match the current credentials' user (compared case-insensitively, "
-                "ignoring surrounding whitespace; see get_default_cloud_context); "
+                "ignoring surrounding whitespace; see get_default_workspace_context); "
                 "mismatches fail with a validation error. "
                 "Required as a safety confirmation."
             ),
@@ -3003,10 +3004,10 @@ def set_default_cloud_workspace(
 
     WARNING: This is a persistent, account-level change. It updates the user's
     stored default workspace in Airbyte Cloud, which affects both future MCP
-    sessions (default_workspace_id in get_default_cloud_context and every tool
+    sessions (default_workspace_id in get_default_workspace_context and every tool
     that falls back to the default workspace) AND the Airbyte Cloud web app,
     where this workspace becomes the user's default landing workspace.
-    Call get_default_cloud_context first to confirm the current user and to
+    Call get_default_workspace_context first to confirm the current user and to
     discover member workspaces.
     """
     result: CloudDefaultWorkspaceUpdateInfo = _get_cloud_client(ctx).set_default_workspace_for_user(
@@ -3031,7 +3032,7 @@ def set_default_cloud_workspace(
     read_only=True,
     idempotent=True,
 )
-def list_cloud_organizations(
+def list_organizations(
     ctx: Context,
     name_contains: Annotated[
         str | None,
@@ -3098,7 +3099,7 @@ def list_cloud_organizations(
             message=(
                 "No organizations were returned for these credentials. Verify the credentials "
                 "or ask the user to provide an organization ID. Call "
-                "`get_default_cloud_context` to inspect your memberships."
+                "`get_default_workspace_context` to inspect your memberships."
             ),
         )
 
@@ -3125,7 +3126,7 @@ def list_cloud_organizations(
     read_only=True,
     idempotent=True,
 )
-def describe_cloud_workspace(
+def describe_workspace(
     ctx: Context,
     *,
     workspace_id: Annotated[
@@ -3142,7 +3143,7 @@ def describe_cloud_workspace(
 ) -> CloudWorkspaceResult:
     """Get basic details about a workspace (ID, name, URL, parent organization).
 
-    Does not include billing/account status; use `get_cloud_organization_billing_status` for that.
+    Does not include billing/account status; use `get_organization_billing_status` for that.
     """
     workspace = _get_cloud_workspace(ctx, workspace_id)
     workspace_response = api_util.get_workspace(
@@ -3170,7 +3171,7 @@ def describe_cloud_workspace(
     read_only=True,
     idempotent=True,
 )
-def describe_cloud_organization(
+def describe_organization(
     ctx: Context,
     *,
     organization_id: Annotated[
@@ -3191,7 +3192,7 @@ def describe_cloud_organization(
 ) -> CloudOrganizationResult:
     """Get basic details about an organization (ID, name, email, feature flags).
 
-    Billing/account status is available via `get_cloud_organization_billing_status`.
+    Billing/account status is available via `get_organization_billing_status`.
 
     With no arguments, resolves the organization from the configured default or the
     authenticated user's sole membership. With multiple memberships, the error lists
@@ -3216,7 +3217,7 @@ def describe_cloud_organization(
     read_only=True,
     idempotent=True,
 )
-def get_cloud_organization_billing_status(
+def get_organization_billing_status(
     ctx: Context,
     *,
     organization_id: Annotated[
@@ -3753,7 +3754,7 @@ def permanently_delete_custom_source_definition(
     destructive=True,
     extra_help_text=DELETE_NAME_GUARD_TIP_TEXT,
 )
-def permanently_delete_cloud_connector(
+def permanently_delete_connector(
     ctx: Context,
     connector_id: Annotated[
         str,
@@ -3827,7 +3828,7 @@ def permanently_delete_cloud_connector(
     destructive=True,
     extra_help_text=DELETE_NAME_GUARD_TIP_TEXT,
 )
-def permanently_delete_cloud_connection(
+def permanently_delete_connection(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -3903,7 +3904,7 @@ def permanently_delete_cloud_connection(
 
 
 @mcp_tool()
-def rename_cloud_connector(
+def rename_connector(
     ctx: Context,
     connector_id: Annotated[
         str,
@@ -3944,7 +3945,7 @@ def rename_cloud_connector(
 @mcp_tool(
     destructive=True,
 )
-def update_cloud_connector_config(
+def update_connector_config(
     ctx: Context,
     connector_id: Annotated[
         str,
@@ -4004,7 +4005,7 @@ def update_cloud_connector_config(
 
 
 @mcp_tool()
-def rename_cloud_connection(
+def rename_connection(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -4038,7 +4039,7 @@ def rename_cloud_connection(
 @mcp_tool(
     destructive=True,
 )
-def set_cloud_connection_table_prefix(
+def set_connection_table_prefix(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -4077,7 +4078,7 @@ def set_cloud_connection_table_prefix(
 @mcp_tool(
     destructive=True,
 )
-def set_cloud_connection_selected_streams(
+def set_connection_selected_streams(
     ctx: Context,
     connection_id: Annotated[
         str,
@@ -4124,7 +4125,7 @@ def set_cloud_connection_selected_streams(
 @mcp_tool(
     destructive=True,
 )
-def update_cloud_connection(
+def update_connection(
     ctx: Context,
     connection_id: Annotated[
         str,
diff --git a/airbyte/mcp/connector_builder.py b/airbyte/mcp/connector_builder.py
index ce6bde917..5338afa01 100644
--- a/airbyte/mcp/connector_builder.py
+++ b/airbyte/mcp/connector_builder.py
@@ -912,7 +912,7 @@ def run_connector_builder_readiness_checks(
             description=(
                 "Optional. If no named stream failed or is still pending (or none are named), "
                 "publish the draft and start a full sync of every stream to a no-op destination. "
-                "Returns the job in `full_sync`; poll it with `get_cloud_sync_status`."
+                "Returns the job in `full_sync`; poll it with `get_connection_sync_status`."
             ),
             default=False,
         ),
diff --git a/airbyte/mcp/guidance.py b/airbyte/mcp/guidance.py
index 4e4e0dc76..93d8bdfc1 100644
--- a/airbyte/mcp/guidance.py
+++ b/airbyte/mcp/guidance.py
@@ -188,7 +188,7 @@ def get_external_access_skill_docs(
         str | None,
         Field(
             description=(
-                "Fully-qualified skill ID, e.g. from `describe_cloud_connector` `skill_id`. "
+                "Fully-qualified skill ID, e.g. from `describe_connector` `skill_id`. "
                 "Provide this or `connector_id`."
             ),
             default=None,
@@ -235,7 +235,7 @@ def get_external_access_skill_docs(
 
     Exactly one of `connector_id` or `docs_skill_id` is required. `connector_id` must be
     the ID of a source or destination deployed in the workspace (for example from
-    `list_cloud_connectors`), not a connector definition ID from the registry.
+    `list_connectors`), not a connector definition ID from the registry.
 
     When direct access is disabled for the connector, `errors` holds a message with an
     enablement link to show the user.
@@ -271,7 +271,7 @@ class ConnectorInfo(BaseModel):
     read_only=True,
     idempotent=True,
 )
-def get_connector_info(
+def get_connector_definition(
     connector_name: Annotated[
         str,
         Field(description="The name of the connector to get information for."),
@@ -281,8 +281,8 @@ def get_connector_info(
     """Get metadata, documentation URL, config spec, and manifest URL for a connector.
 
     This searches public registry connector types, not workspace-deployed sources, destinations,
-    or connections; use `list_cloud_connectors`, `list_cloud_connections`, or
-    `describe_cloud_connection` for workspace resources. For custom Builder/YAML connectors, use
+    or connections; use `list_connectors`, `list_connections`, or
+    `describe_connection` for workspace resources. For custom Builder/YAML connectors, use
     `list_custom_source_definitions` or `get_custom_source_definition`.
 
     `config_spec_jsonschema` is fetched from the public connector registry over
diff --git a/airbyte/mcp/interactive/__init__.py b/airbyte/mcp/interactive/__init__.py
index 0cdd75dd4..cdad6670d 100644
--- a/airbyte/mcp/interactive/__init__.py
+++ b/airbyte/mcp/interactive/__init__.py
@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING
 
 from airbyte.mcp._tool_utils import register_mcp_tools
-from airbyte.mcp.interactive._registry_ui import show_connectors_list
+from airbyte.mcp.interactive._registry_ui import show_connector_definitions
 from airbyte.mcp.interactive._sync_history_ui import show_connection_sync_history
 from airbyte.mcp.interactive._workspace_sync_status_ui import show_workspace_sync_status
 
@@ -25,7 +25,7 @@ def register_interactive_tools(app: FastMCP) -> None:
 
 __all__ = [
     "register_interactive_tools",
-    "show_connectors_list",
+    "show_connector_definitions",
     "show_connection_sync_history",
     "show_workspace_sync_status",
 ]
diff --git a/airbyte/mcp/interactive/_registry_ui.py b/airbyte/mcp/interactive/_registry_ui.py
index 0dbb371c4..7314ec4e1 100644
--- a/airbyte/mcp/interactive/_registry_ui.py
+++ b/airbyte/mcp/interactive/_registry_ui.py
@@ -45,7 +45,7 @@
     PublicConnectorSummary,
     SupportLevel,
 )
-from airbyte.mcp.registry import list_connectors as _list_connectors
+from airbyte.mcp.registry import list_connector_definitions as _list_connector_definitions
 from airbyte.registry import ConnectorMetadata, _get_registry_url, get_connector_metadata
 
 
@@ -83,7 +83,7 @@ def _json_dumps(value: JsonValue) -> str:
     idempotent=True,
     app=PrefabAppConfig(),
 )
-def show_connectors_list(
+def show_connector_definitions(
     support_level: Annotated[
         str,
         Field(
@@ -385,7 +385,7 @@ def _list_public_registry_connectors(
     search: str = "",
     limit: int | None = None,
 ) -> list[PublicConnectorSummary]:
-    connector_names = _list_connectors(
+    connector_names = _list_connector_definitions(
         keyword_filter=None,
         connector_type_filter=_connector_type_value(connector_type),
         install_types=None,
diff --git a/airbyte/mcp/local.py b/airbyte/mcp/local.py
index cee0f6d49..55a735aad 100644
--- a/airbyte/mcp/local.py
+++ b/airbyte/mcp/local.py
@@ -141,7 +141,7 @@ def _get_mcp_source(
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def validate_connector_config(
+def validate_local_connector_config(
     connector_name: Annotated[
         str,
         Field(description="The name of the connector to validate."),
@@ -231,7 +231,7 @@ def validate_connector_config(
     idempotent=True,
     requires_client_filesystem=True,
 )
-def list_connector_config_secrets(
+def list_local_connector_config_secrets(
     connector_name: Annotated[
         str,
         Field(description="The name of the connector."),
@@ -245,7 +245,7 @@ def list_connector_config_secrets(
     return the actual secret values.
     """
     raise_if_untrusted_execution_context(
-        "Listing connector config secrets (`list_connector_config_secrets`)"
+        "Listing connector config secrets (`list_local_connector_config_secrets`)"
     )
     secrets_names: list[str] = []
     for secrets_mgr in _get_secret_sources():
@@ -265,13 +265,15 @@ def list_connector_config_secrets(
     idempotent=True,
     requires_client_filesystem=True,
 )
-def list_dotenv_secrets() -> dict[str, list[str]]:
+def list_local_dotenv_secrets() -> dict[str, list[str]]:
     """List all environment variable names declared within declared .env files.
 
     This returns a dictionary mapping the .env file name to a list of environment
     variable names. The values of the environment variables are not returned.
     """
-    raise_if_untrusted_execution_context("Listing dotenv secret names (`list_dotenv_secrets`)")
+    raise_if_untrusted_execution_context(
+        "Listing dotenv secret names (`list_local_dotenv_secrets`)"
+    )
     result: dict[str, list[str]] = {}
     for secrets_mgr in _get_secret_sources():
         if isinstance(secrets_mgr, DotenvSecretManager) and secrets_mgr.dotenv_path:
@@ -286,7 +288,7 @@ def list_dotenv_secrets() -> dict[str, list[str]]:
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def list_source_streams(
+def list_local_source_streams(
     source_connector_name: Annotated[
         str,
         Field(description="The name of the source connector."),
@@ -357,7 +359,7 @@ def list_source_streams(
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def get_source_stream_json_schema(
+def get_local_source_stream_json_schema(
     source_connector_name: Annotated[
         str,
         Field(description="The name of the source connector."),
@@ -429,7 +431,7 @@ def get_source_stream_json_schema(
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def read_source_stream_records(
+def read_local_source_stream_records(
     source_connector_name: Annotated[
         str,
         Field(description="The name of the source connector."),
@@ -531,7 +533,7 @@ def read_source_stream_records(
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def get_stream_previews(
+def get_local_stream_previews(
     source_name: Annotated[
         str,
         Field(description="The name of the source connector."),
@@ -655,7 +657,7 @@ def get_stream_previews(
     requires_client_filesystem=True,
     extra_help_text=_CONFIG_HELP,
 )
-def sync_source_to_cache(
+def sync_local_source_to_cache(
     source_connector_name: Annotated[
         str,
         Field(description="The name of the source connector."),
@@ -769,9 +771,11 @@ class CachedDatasetInfo(BaseModel):
     idempotent=True,
     requires_client_filesystem=True,
 )
-def list_cached_streams() -> list[CachedDatasetInfo]:
+def list_local_cached_streams() -> list[CachedDatasetInfo]:
     """List all streams available in the default DuckDB cache."""
-    raise_if_untrusted_execution_context("Reading the local default cache (`list_cached_streams`)")
+    raise_if_untrusted_execution_context(
+        "Reading the local default cache (`list_local_cached_streams`)"
+    )
     cache: DuckDBCache = get_default_cache()
     result = [
         CachedDatasetInfo(
@@ -790,10 +794,10 @@ def list_cached_streams() -> list[CachedDatasetInfo]:
     idempotent=True,
     requires_client_filesystem=True,
 )
-def describe_default_cache() -> dict[str, Any]:
+def describe_local_default_cache() -> dict[str, Any]:
     """Describe the currently configured default cache."""
     raise_if_untrusted_execution_context(
-        "Describing the local default cache (`describe_default_cache`)"
+        "Describing the local default cache (`describe_local_default_cache`)"
     )
     cache = get_default_cache()
     return {
@@ -844,7 +848,7 @@ def _is_safe_sql(sql_query: str) -> bool:
     idempotent=True,
     requires_client_filesystem=True,
 )
-def run_sql_query(
+def run_local_sql_query(
     sql_query: Annotated[
         str,
         Field(description="The SQL query to execute."),
@@ -860,7 +864,7 @@ def run_sql_query(
     """Run a SQL query against the default cache.
 
     The dialect of SQL should match the dialect of the default cache.
-    Use `describe_default_cache` to see the cache type.
+    Use `describe_local_default_cache` to see the cache type.
 
     For DuckDB-type caches:
     - Use `SHOW TABLES` to list all tables.
@@ -868,7 +872,7 @@ def run_sql_query(
 
     For security reasons, only read-only operations are allowed: SELECT, DESCRIBE, SHOW, EXPLAIN.
     """
-    raise_if_untrusted_execution_context("Querying the local default cache (`run_sql_query`)")
+    raise_if_untrusted_execution_context("Querying the local default cache (`run_local_sql_query`)")
     # Check if the query is safe to execute
     if not _is_safe_sql(sql_query):
         return [
@@ -905,7 +909,7 @@ def run_sql_query(
     destructive=True,
     requires_client_filesystem=True,
 )
-def destination_smoke_test(  # noqa: PLR0913, PLR0917
+def run_local_destination_smoke_test(  # noqa: PLR0913, PLR0917
     destination_connector_name: Annotated[
         str,
         Field(
@@ -1028,7 +1032,9 @@ def destination_smoke_test(  # noqa: PLR0913, PLR0917
     per-column null/non-null counts. Results are included in the response
     as `table_statistics` and `tables_not_found`.
     """
-    raise_if_untrusted_execution_context("Destination smoke test (`destination_smoke_test`)")
+    raise_if_untrusted_execution_context(
+        "Destination smoke test (`run_local_destination_smoke_test`)"
+    )
     # Resolve destination config
     config_dict = resolve_connector_config(
         config=config,
diff --git a/airbyte/mcp/registry.py b/airbyte/mcp/registry.py
index ceb9b56a5..0962c8270 100644
--- a/airbyte/mcp/registry.py
+++ b/airbyte/mcp/registry.py
@@ -37,7 +37,7 @@
     read_only=True,
     idempotent=True,
 )
-def list_connectors(
+def list_connector_definitions(
     keyword_filter: Annotated[
         str | None,
         Field(
@@ -80,11 +80,11 @@ def list_connectors(
         ),
     ],
 ) -> list[str]:
-    """List available Airbyte connectors with optional filtering.
+    """List connector types available to deploy (the Airbyte connector registry).
 
-    This lists connector types in the public registry, not sources, destinations, or connections
-    deployed in a workspace; use `list_cloud_connectors`, `list_cloud_connections`, or
-    `describe_cloud_connection` for workspace resources.
+    Use `list_connectors` for deployed sources and destinations in a workspace, and
+    `list_custom_source_definitions` for workspace custom definitions. Use `list_connections`
+    or `describe_connection` for workspace pipelines. Optional filters narrow the registry.
 
     Returns:
         List of connector names.
@@ -129,7 +129,7 @@ def list_connectors(
     read_only=True,
     idempotent=True,
 )
-def get_connector_version_history(
+def get_connector_definition_version_history(
     connector_name: Annotated[
         str,
         Field(
diff --git a/airbyte/mcp/server.py b/airbyte/mcp/server.py
index 7ff5bd9de..3fcf4585c 100644
--- a/airbyte/mcp/server.py
+++ b/airbyte/mcp/server.py
@@ -172,8 +172,8 @@
 _INSTRUCTIONS_WORKSPACE_GUIDANCE = """
   When a tool's workspace_id is omitted, the session's workspace is used: the
   workspace configured for the connection if one is set, otherwise the
-  authenticated user's default (or only) workspace. Use get_default_cloud_context or
-  list_cloud_workspaces to discover workspaces. Only call list_cloud_organizations
+  authenticated user's default (or only) workspace. Use get_default_workspace_context or
+  list_workspaces to discover workspaces. Only call list_organizations
   when you need to search organizations by name, passing name_contains. If multiple
   organizations or workspaces are candidates, ask the user to choose; never select
   automatically."""
diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md
index e2260f751..0c61d9f48 100644
--- a/docs/CONTRIBUTING.md
+++ b/docs/CONTRIBUTING.md
@@ -122,15 +122,15 @@ The easiest way to test Airbyte Replication MCP tools during development is usin
 ```bash
 poe mcp-tool-test <tool_name> '<json_args>'
 
-poe mcp-tool-test list_connectors '{}'
+poe mcp-tool-test list_connector_definitions '{}'
 poe mcp-tool-test get_config_spec '{"connector_name": "source-pokeapi"}'
 poe mcp-tool-test validate_config \
     '{"connector_name": "source-pokeapi", "config": {"pokemon_name": "pikachu"}}'
 poe mcp-tool-test run_sync \
     '{"connector_name": "source-pokeapi", "config": {"pokemon_name": "pikachu"}}'
 
-poe mcp-tool-test get_default_cloud_context '{}'
-poe mcp-tool-test list_cloud_connections '{}'
+poe mcp-tool-test get_default_workspace_context '{}'
+poe mcp-tool-test list_connections '{}'
 ```
 
 You can also invoke the server using one of these helper tasks:
@@ -174,7 +174,7 @@ PyAirbyte server that is:
 Inside each module page, primitives are grouped by kind (`## Tools`,
 `## Prompts`, `## Resources`), and each primitive has an HTML anchor
 (`<a id="name"></a>`) above its H3 so links like
-`cloud.md#deploy_connector_to_cloud` resolve in both pdoc and Docusaurus.
+`cloud.md#deploy_connector` resolve in both pdoc and Docusaurus.
 Regenerate after any change to MCP tool signatures, descriptions, or schemas.
 
 ## Removing Environment Variables and MCP Headers
diff --git a/scripts/generate_mcp_markdown.py b/scripts/generate_mcp_markdown.py
index dd5d882e4..e55a184a9 100755
--- a/scripts/generate_mcp_markdown.py
+++ b/scripts/generate_mcp_markdown.py
@@ -25,7 +25,7 @@
 # cloud module
 
 ## Tools (35)
-### deploy_connector_to_cloud
+### deploy_connector
 ...
 ## Prompts (N)
 ### some_prompt
@@ -50,7 +50,7 @@
   components are used.
 - **Deep-linkable**: every tool/resource/prompt name gets an HTML anchor
   (`<a id="name"></a>`) above its H3, so links like
-  `cloud.md#deploy_connector_to_cloud` resolve in both pdoc and Docusaurus.
+  `cloud.md#deploy_connector` resolve in both pdoc and Docusaurus.
 
 Formatting is modeled on the
 [`mcpdocs-gen`](https://github.com/smytsyk/mcpdocs) static HTML output — same
diff --git a/tests/unit_tests/fixtures/arg_trace_plan.json b/tests/unit_tests/fixtures/arg_trace_plan.json
index dfc5c2f55..f4fc53a87 100644
--- a/tests/unit_tests/fixtures/arg_trace_plan.json
+++ b/tests/unit_tests/fixtures/arg_trace_plan.json
@@ -1,15 +1,15 @@
 {
-  "cancel_cloud_sync": {
+  "cancel_connection_sync": {
     "connection_id": "hash",
     "job_id": "hash",
     "workspace_id": "omit"
   },
-  "check_cloud_connector": {
+  "check_connector": {
     "connector_id": "hash",
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "workspace_id": "omit"
   },
-  "create_connection_on_cloud": {
+  "create_connection": {
     "connection_name": "fingerprint",
     "destination_id": "hash",
     "selected_streams": "fingerprint list",
@@ -25,7 +25,7 @@
     "pre_validate": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "deploy_connector_to_cloud": {
+  "deploy_connector": {
     "config": "presence",
     "config_secret_name": "fingerprint",
     "connector_name": "fingerprint",
@@ -36,16 +36,16 @@
     "unique": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "deploy_noop_destination_to_cloud": {
+  "deploy_noop_destination": {
     "destination_name": "fingerprint",
     "unique": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "describe_cloud_connection": {
+  "describe_connection": {
     "connection_id": "hash",
     "workspace_id": "omit"
   },
-  "describe_cloud_connector": {
+  "describe_connector": {
     "connector_id": "hash",
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "with_config": "value allowed=[false,true]",
@@ -54,15 +54,15 @@
     "with_replication_details": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "describe_cloud_organization": {
+  "describe_organization": {
     "organization_id": "hash",
     "organization_name": "fingerprint"
   },
-  "describe_cloud_workspace": {
+  "describe_workspace": {
     "workspace_id": "omit"
   },
-  "describe_default_cache": {},
-  "destination_smoke_test": {
+  "describe_local_default_cache": {},
+  "run_local_destination_smoke_test": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -113,18 +113,18 @@
   "get_api_docs_urls": {
     "connector_name": "fingerprint"
   },
-  "get_cloud_organization_billing_status": {
+  "get_organization_billing_status": {
     "organization_id": "hash",
     "organization_name": "fingerprint"
   },
-  "get_cloud_search_status": {
+  "get_search_index_status": {
     "connector_id": "hash",
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "namespace": "fingerprint",
     "stream_name": "fingerprint",
     "workspace_id": "omit"
   },
-  "get_cloud_sync_logs": {
+  "get_connection_sync_logs": {
     "attempt_number": "hash",
     "connection_id": "hash",
     "from_tail": "value allowed=[false,true]",
@@ -133,7 +133,7 @@
     "max_lines": "hash",
     "workspace_id": "omit"
   },
-  "get_cloud_sync_status": {
+  "get_connection_sync_status": {
     "connection_id": "hash",
     "include_attempts": "value allowed=[false,true]",
     "job_id": "hash",
@@ -161,10 +161,10 @@
     "stream_limit": "value int",
     "workspace_id": "omit"
   },
-  "get_connector_info": {
+  "get_connector_definition": {
     "connector_name": "fingerprint"
   },
-  "get_connector_version_history": {
+  "get_connector_definition_version_history": {
     "connector_name": "fingerprint",
     "limit": "value int",
     "num_versions_to_validate": "hash"
@@ -174,14 +174,14 @@
     "include_draft": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "get_default_cloud_context": {},
+  "get_default_workspace_context": {},
   "get_external_access_skill_docs": {
     "connector_id": "hash",
     "docs_skill_id": "hash",
     "section": "fingerprint",
     "workspace_id": "omit"
   },
-  "get_source_stream_json_schema": {
+  "get_local_source_stream_json_schema": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -190,7 +190,7 @@
     "source_connector_name": "fingerprint",
     "stream_name": "fingerprint"
   },
-  "get_stream_previews": {
+  "get_local_stream_previews": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -200,34 +200,34 @@
     "source_name": "fingerprint",
     "streams": "fingerprint list"
   },
-  "list_cached_streams": {},
-  "list_cloud_connections": {
+  "list_local_cached_streams": {},
+  "list_connections": {
     "failing_connections_only": "value allowed=[false,true]",
     "limit": "value int",
     "name_contains": "fingerprint",
     "with_connection_status": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "list_cloud_connectors": {
+  "list_connectors": {
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "feature_filter": "value allowed=[\"direct_access\",\"direct_api_action\",\"direct_api_query\",\"direct_sql_query\",\"search_indexing\"]",
     "limit": "value int",
     "name_contains": "fingerprint",
     "workspace_id": "omit"
   },
-  "list_cloud_organizations": {
+  "list_organizations": {
     "feature_filter": "value allowed=[\"direct_access\",\"search_indexing\"]",
     "limit": "value int",
     "name_contains": "fingerprint"
   },
-  "list_cloud_sync_jobs": {
+  "list_connection_sync_jobs": {
     "connection_id": "hash",
     "from_tail": "value allowed=[false,true]",
     "job_type": "value allowed=[\"clear\",\"refresh\",\"reset\",\"sync\"]",
     "max_jobs": "hash",
     "workspace_id": "omit"
   },
-  "list_cloud_workspaces": {
+  "list_workspaces": {
     "limit": "value int",
     "name_contains": "fingerprint",
     "organization_id": "hash",
@@ -239,10 +239,10 @@
     "definition_id": "hash",
     "workspace_id": "omit"
   },
-  "list_connector_config_secrets": {
+  "list_local_connector_config_secrets": {
     "connector_name": "fingerprint"
   },
-  "list_connectors": {
+  "list_connector_definitions": {
     "connector_type_filter": "value allowed=[\"destination\",\"source\"]",
     "install_types": "value allowed=[\"docker\",\"java\",\"python\",\"yaml\"] list",
     "keyword_filter": "fingerprint"
@@ -250,8 +250,8 @@
   "list_custom_source_definitions": {
     "workspace_id": "omit"
   },
-  "list_dotenv_secrets": {},
-  "list_source_streams": {
+  "list_local_dotenv_secrets": {},
+  "list_local_source_streams": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -259,14 +259,14 @@
     "override_execution_mode": "value allowed=[\"auto\",\"docker\",\"python\",\"yaml\"]",
     "source_connector_name": "fingerprint"
   },
-  "permanently_delete_cloud_connection": {
+  "permanently_delete_connection": {
     "cascade_delete_destination": "value allowed=[false,true]",
     "cascade_delete_source": "value allowed=[false,true]",
     "connection_id": "hash",
     "name": "fingerprint",
     "workspace_id": "omit"
   },
-  "permanently_delete_cloud_connector": {
+  "permanently_delete_connector": {
     "connector_id": "hash",
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "name": "fingerprint",
@@ -288,7 +288,7 @@
     "unique": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "read_source_stream_records": {
+  "read_local_source_stream_records": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -298,18 +298,18 @@
     "source_connector_name": "fingerprint",
     "stream_name": "fingerprint"
   },
-  "rename_cloud_connection": {
+  "rename_connection": {
     "connection_id": "hash",
     "name": "fingerprint",
     "workspace_id": "omit"
   },
-  "rename_cloud_connector": {
+  "rename_connector": {
     "connector_id": "hash",
     "connector_type": "value allowed=[\"destination\",\"source\"]",
     "name": "fingerprint",
     "workspace_id": "omit"
   },
-  "request_cloud_search_index_backfill": {
+  "request_search_index_backfill": {
     "backfill_start_time": "value str",
     "connection_id": "hash",
     "streams": "hash list",
@@ -323,7 +323,7 @@
     "version": "value int",
     "workspace_id": "omit"
   },
-  "run_cloud_sync": {
+  "run_connection_sync": {
     "connection_id": "hash",
     "wait": "value allowed=[false,true]",
     "wait_timeout": "hash",
@@ -347,19 +347,19 @@
     "stream_name": "fingerprint",
     "workspace_id": "omit"
   },
-  "run_sql_query": {
+  "run_local_sql_query": {
     "max_records": "hash",
     "sql_query": "hash"
   },
   "search_airbyte_knowledge_sources": {
     "query": "fingerprint"
   },
-  "set_cloud_connection_selected_streams": {
+  "set_connection_selected_streams": {
     "connection_id": "hash",
     "stream_names": "fingerprint list",
     "workspace_id": "omit"
   },
-  "set_cloud_connection_table_prefix": {
+  "set_connection_table_prefix": {
     "connection_id": "hash",
     "prefix": "fingerprint",
     "workspace_id": "omit"
@@ -370,7 +370,7 @@
     "nonsecret_testing_values": "presence",
     "workspace_id": "omit"
   },
-  "set_default_cloud_workspace": {
+  "set_default_workspace": {
     "user_email": "hash",
     "workspace_id": "omit"
   },
@@ -381,7 +381,7 @@
     "suppress_ui": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "show_connectors_list": {
+  "show_connector_definitions": {
     "certified": "value allowed=[false,true]",
     "connector_type": "value str",
     "limit": "value int",
@@ -397,7 +397,7 @@
     "suppress_ui": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "sync_source_to_cache": {
+  "sync_local_source_to_cache": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
@@ -406,7 +406,7 @@
     "source_connector_name": "fingerprint",
     "streams": "fingerprint list"
   },
-  "update_cloud_connection": {
+  "update_connection": {
     "connection_id": "hash",
     "cron_expression": "fingerprint",
     "enabled": "value allowed=[false,true]",
@@ -414,7 +414,7 @@
     "manual_schedule": "value allowed=[false,true]",
     "workspace_id": "omit"
   },
-  "update_cloud_connector_config": {
+  "update_connector_config": {
     "config": "presence",
     "config_secret_name": "fingerprint",
     "connector_id": "hash",
@@ -440,7 +440,7 @@
     "testing_values_secret_name": "fingerprint",
     "workspace_id": "omit"
   },
-  "validate_connector_config": {
+  "validate_local_connector_config": {
     "config": "presence",
     "config_file": "fingerprint",
     "config_secret_name": "fingerprint",
diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py
index c8f2ded4f..64ffdf70f 100644
--- a/tests/unit_tests/test_cloud_api_util.py
+++ b/tests/unit_tests/test_cloud_api_util.py
@@ -2399,7 +2399,7 @@ def test_create_connection_400_with_non_json_body_uses_fallback_message(
             json.dumps({"title": "actor-not-ready"}),
             True,
             AirbyteConnectorNotReadyError,
-            "`check_cloud_connector`",
+            "`check_connector`",
             id="actor-not-ready-mcp",
         ),
         pytest.param(
@@ -2415,7 +2415,7 @@ def test_create_connection_400_with_non_json_body_uses_fallback_message(
             }),
             True,
             AirbyteConnectorNotReadyError,
-            "`check_cloud_connector`",
+            "`check_connector`",
             id="actor-not-ready-by-type",
         ),
         pytest.param(
diff --git a/tests/unit_tests/test_cloud_client.py b/tests/unit_tests/test_cloud_client.py
index 0c6c172e2..9d528d210 100644
--- a/tests/unit_tests/test_cloud_client.py
+++ b/tests/unit_tests/test_cloud_client.py
@@ -1036,9 +1036,7 @@ def test_get_default_context_for_user_is_bounded_to_permission_derived_scope() -
     assert context.member_organizations_truncated is False
     assert context.member_workspaces_truncated is False
     assert len(context.discovery_hints) == 3
-    assert any(
-        "set_default_cloud_workspace" in hint for hint in context.discovery_hints
-    )
+    assert any("set_default_workspace" in hint for hint in context.discovery_hints)
 
 
 def test_get_default_context_for_user_truncates_organization_memberships() -> None:
@@ -1152,7 +1150,7 @@ def test_get_default_context_for_user_hints_setter_when_stored_default_missing()
     assert context.default_workspace_id == "workspace-1"
     assert context.discovery_hints == [
         "No default workspace is set. Use "
-        "set_default_cloud_workspace(user_email=<your email>, "
+        "set_default_workspace(user_email=<your email>, "
         "workspace_id=<id>) to durably set one; it applies to both MCP "
         "sessions and the Airbyte Cloud web app."
     ]
@@ -1168,7 +1166,7 @@ def test_get_default_context_for_user_verified_is_none_without_default() -> None
     assert context.default_workspace_verified is None
     assert context.discovery_hints == [
         "No default workspace is set. Use "
-        "set_default_cloud_workspace(user_email=<your email>, "
+        "set_default_workspace(user_email=<your email>, "
         "workspace_id=<id>) to durably set one; it applies to both MCP "
         "sessions and the Airbyte Cloud web app."
     ]
@@ -1203,7 +1201,7 @@ def test_get_default_context_for_user_configured_default_hint() -> None:
     assert context.discovery_hints == [
         "Your Airbyte user profile has no default workspace; this session "
         "uses the configured workspace configured-workspace. Use "
-        "set_default_cloud_workspace(user_email=<your email>, "
+        "set_default_workspace(user_email=<your email>, "
         "workspace_id=<id>) to set a durable default; it applies to "
         "both MCP sessions and the Airbyte Cloud web app."
     ]
@@ -1371,7 +1369,7 @@ def test_set_default_workspace_for_user_rejects_email_mismatch() -> None:
 
     assert "does not match the authenticated user" in str(exc_info.value)
     assert "user@example.com" in str(exc_info.value.context)
-    assert "get_default_cloud_context" in str(exc_info.value.guidance)
+    assert "get_default_workspace_context" in str(exc_info.value.guidance)
     get_workspace.assert_not_called()
     update_user.assert_not_called()
 
diff --git a/tests/unit_tests/test_cloud_connections.py b/tests/unit_tests/test_cloud_connections.py
index c609b5ec6..68dd83460 100644
--- a/tests/unit_tests/test_cloud_connections.py
+++ b/tests/unit_tests/test_cloud_connections.py
@@ -439,7 +439,7 @@ def test_get_sync_result_validates_job_id(
         assert exc_info.value.resource_type == "sync job"
         assert exc_info.value.resource_name_or_id == "42"
         assert exc_info.value.guidance == (
-            "Use `list_cloud_sync_jobs` to find valid job IDs for this connection."
+            "Use `list_connection_sync_jobs` to find valid job IDs for this connection."
         )
         assert exc_info.value.context == {
             "connection_id": "connection-id",
@@ -460,7 +460,7 @@ def test_get_sync_result_validates_job_id(
         assert exc_info.value.resource_type == "sync job"
         assert exc_info.value.resource_name_or_id == "42"
         assert exc_info.value.guidance == (
-            "Use `list_cloud_sync_jobs` to find valid job IDs for this connection."
+            "Use `list_connection_sync_jobs` to find valid job IDs for this connection."
         )
         assert exc_info.value.context == {
             "connection_id": "connection-id",
diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py
index e78df88a5..79b28abeb 100644
--- a/tests/unit_tests/test_cloud_credentials.py
+++ b/tests/unit_tests/test_cloud_credentials.py
@@ -1810,7 +1810,7 @@ def get_organization(
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.describe_cloud_organization(
+    result = mcp_cloud.describe_organization(
         None,
         organization_id=None,
         organization_name=None,
@@ -1823,7 +1823,7 @@ def get_organization(
 def test_mcp_describe_cloud_organization_returns_email_and_domain(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """`describe_cloud_organization` reports the contact email and its domain."""
+    """`describe_organization` reports the contact email and its domain."""
     _stub_organization_features(monkeypatch)
 
     class DiscoveryClient:
@@ -1836,7 +1836,7 @@ def get_organization(self, **_: object) -> CloudOrganization:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.describe_cloud_organization(
+    result = mcp_cloud.describe_organization(
         None, organization_id="organization-id", organization_name=None
     )
 
@@ -1847,7 +1847,7 @@ def get_organization(self, **_: object) -> CloudOrganization:
 def test_mcp_list_cloud_organizations_hides_email_and_reports_domain(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """`list_cloud_organizations` omits contact emails and reports only the domain."""
+    """`list_organizations` omits contact emails and reports only the domain."""
     _stub_organization_features(monkeypatch)
 
     class DiscoveryClient:
@@ -1862,7 +1862,7 @@ def list_organizations(self, **_: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(None)
+    result = mcp_cloud.list_organizations(None)
 
     assert result.organizations[0].email is None
     assert result.organizations[0].email_domain == "example.com"
@@ -2004,7 +2004,7 @@ def list_organizations(self, **_: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(None)
+    result = mcp_cloud.list_organizations(None)
 
     assert len(result.organizations) == expected_count
     if expected_message is not None:
@@ -2024,7 +2024,7 @@ def list_organizations(self, **_: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(None)
+    result = mcp_cloud.list_organizations(None)
 
     assert result.organizations == [
         mcp_cloud.CloudOrganizationResult(
@@ -2732,7 +2732,7 @@ def test_mcp_list_cloud_connectors_enablement_failure_is_unknown(
     monkeypatch: pytest.MonkeyPatch,
     error: Exception,
 ) -> None:
-    """Enablement failures degrade to `"unknown"` in `list_cloud_connectors`, never raise."""
+    """Enablement failures degrade to `"unknown"` in `list_connectors`, never raise."""
     workspace = _make_workspace(
         monkeypatch, organization_info={"organizationId": "organization-id"}
     )
@@ -2755,7 +2755,7 @@ def fail_for_source_2(**kwargs: object) -> ConnectorEnablement:
     )
     monkeypatch.setattr(mcp_cloud, "_get_cloud_workspace", lambda _ctx, _id: workspace)
 
-    results = mcp_cloud.list_cloud_connectors(
+    results = mcp_cloud.list_connectors(
         None,
         connector_type=ConnectorType.SOURCE,
         workspace_id=None,
@@ -2792,7 +2792,7 @@ def test_mcp_describe_cloud_connector_enablement_failure_is_unknown(
     monkeypatch: pytest.MonkeyPatch,
     error: Exception,
 ) -> None:
-    """Enablement failures degrade to `"unknown"` in `describe_cloud_connector`."""
+    """Enablement failures degrade to `"unknown"` in `describe_connector`."""
     workspace = _make_workspace(
         monkeypatch, organization_info={"organizationId": "organization-id"}
     )
@@ -2906,7 +2906,7 @@ def test_mcp_list_cloud_connectors_source_features(
     )
     monkeypatch.setattr(mcp_cloud, "_get_cloud_workspace", lambda _ctx, _id: workspace)
 
-    results = mcp_cloud.list_cloud_connectors(
+    results = mcp_cloud.list_connectors(
         None,
         connector_type=ConnectorType.SOURCE,
         workspace_id=None,
@@ -2956,7 +2956,7 @@ def test_mcp_list_cloud_connectors_destination_features(
     _patch_workspace_connectors(monkeypatch, workspace)
     monkeypatch.setattr(mcp_cloud, "_get_cloud_workspace", lambda _ctx, _id: workspace)
 
-    results = mcp_cloud.list_cloud_connectors(
+    results = mcp_cloud.list_connectors(
         None,
         connector_type=ConnectorType.DESTINATION,
         workspace_id=None,
@@ -2989,7 +2989,7 @@ def test_mcp_list_cloud_connectors_rejects_non_positive_limit(
     monkeypatch.setattr(mcp_cloud, "_get_cloud_workspace", lambda _ctx, _id: workspace)
 
     with pytest.raises(AirbyteLibInputError, match="`limit` must be greater than 0."):
-        mcp_cloud.list_cloud_connectors(
+        mcp_cloud.list_connectors(
             None,
             connector_type=connector_type,
             workspace_id=None,
@@ -3132,7 +3132,7 @@ def get_organization(self, *, organization_id: str) -> CloudOrganization:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_workspaces(
+    result = mcp_cloud.list_workspaces(
         None,
         organization_id=None,
         organization_name=None,
@@ -3179,7 +3179,7 @@ def list_organizations(self, **kwargs: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(
+    result = mcp_cloud.list_organizations(
         None,
         name_contains="develop",
         limit=1,
@@ -3215,7 +3215,7 @@ def list_organizations(self, **_: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(None, feature_filter=feature_filter)
+    result = mcp_cloud.list_organizations(None, feature_filter=feature_filter)
 
     assert result.organizations == []
     assert expected_fragment in (result.message or "")
@@ -3304,7 +3304,7 @@ def list_organizations(self, **kwargs: object) -> list[CloudOrganization]:
 
     monkeypatch.setattr(mcp_cloud, "_get_cloud_client", lambda _: DiscoveryClient())
 
-    result = mcp_cloud.list_cloud_organizations(
+    result = mcp_cloud.list_organizations(
         None, feature_filter=OrganizationFeature.DIRECT_ACCESS
     )
 
diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py
index d4f5be928..a97952035 100644
--- a/tests/unit_tests/test_cloud_direct_execute.py
+++ b/tests/unit_tests/test_cloud_direct_execute.py
@@ -1511,7 +1511,7 @@ def test_search_result_warns_when_no_index_was_searched() -> None:
 
     assert len(result.warnings) == 1
     assert "No index matched" in result.warnings[0]
-    assert "get_cloud_search_status" in result.warnings[0]
+    assert "get_search_index_status" in result.warnings[0]
 
 
 def test_search_status_result_accepts_null_warnings() -> None:
diff --git a/tests/unit_tests/test_deferred_cloud_credentials.py b/tests/unit_tests/test_deferred_cloud_credentials.py
index c7a39fab1..3ccf904ec 100644
--- a/tests/unit_tests/test_deferred_cloud_credentials.py
+++ b/tests/unit_tests/test_deferred_cloud_credentials.py
@@ -771,7 +771,7 @@ def test_mcp_deploy_with_deferred_credentials_returns_handoff(
     connector_type: str,
 ) -> None:
     """Deferred MCP deploys return the settings link and next-step guidance."""
-    raw = cloud_mcp.deploy_connector_to_cloud(
+    raw = cloud_mcp.deploy_connector(
         ctx=cast(Context, object()),
         name="My connector",
         connector_name=f"{connector_type}-faker",
@@ -787,7 +787,7 @@ def test_mcp_deploy_with_deferred_credentials_returns_handoff(
     assert result.connector_type == connector_type
     assert result.workspace_id == WORKSPACE_ID
     assert result.settings_url.endswith("/settings")
-    assert "check_cloud_connector" in result.guidance
+    assert "check_connector" in result.guidance
     assert "`workspace_id`" in result.guidance
     assert "409 actor-not-ready" in result.guidance
     assert "creating a connection" in result.guidance
@@ -816,7 +816,7 @@ def _unacknowledged(**kwargs: Any) -> _DeployedLike:  # noqa: ANN401
     monkeypatch.setattr(workspace_like, "deploy_source", _unacknowledged)
 
     with pytest.raises(exc.AirbyteDeferredSetupError):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             ctx=cast(Context, object()),
             name="My connector",
             connector_name="source-faker",
@@ -835,7 +835,7 @@ def test_mcp_deploy_deferred_rejects_connector_type_mismatch(
 ) -> None:
     """An explicit `connector_type` must agree with the registry's type for the connector."""
     with pytest.raises(exc.AirbyteLibInputError, match="not a source connector"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             ctx=cast(Context, object()),
             name="My connector",
             connector_name="destination-faker",
@@ -867,7 +867,7 @@ def _validated_deploy(**kwargs: Any) -> _DeployedLike:  # noqa: ANN401
     monkeypatch.setattr(workspace_like, "deploy_source", _validated_deploy)
 
     with pytest.raises(exc.AirbyteLibInputError, match="credential values"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             ctx=cast(Context, object()),
             name="My connector",
             connector_name="source-faker",
@@ -881,9 +881,7 @@ def _validated_deploy(**kwargs: Any) -> _DeployedLike:  # noqa: ANN401
 
 def test_mcp_check_cloud_connector_is_read_only_and_idempotent() -> None:
     """The check only reads the connector's check result; it mutates nothing."""
-    (annotations,) = [
-        a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.check_cloud_connector
-    ]
+    (annotations,) = [a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.check_connector]
     assert annotations["readOnlyHint"] is True
     assert annotations["idempotentHint"] is True
 
@@ -893,7 +891,7 @@ def test_mcp_deploy_deferred_rejects_config_secret_name(
 ) -> None:
     """Server-side secrets cannot be combined with deferred credentials."""
     with pytest.raises(exc.AirbyteLibInputError, match="config_secret_name"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             ctx=cast(Context, object()),
             name="My connector",
             connector_name="source-faker",
diff --git a/tests/unit_tests/test_exceptions.py b/tests/unit_tests/test_exceptions.py
index 7777139c6..8c2496aa8 100644
--- a/tests/unit_tests/test_exceptions.py
+++ b/tests/unit_tests/test_exceptions.py
@@ -110,12 +110,12 @@ def test_cloud_credentials_error_guidance(
         pytest.param(
             True,
             "The authenticated user's default workspace was checked and none was "
-            "available. `list_cloud_workspaces` returns direct workspace memberships "
+            "available. `list_workspaces` returns direct workspace memberships "
             "by default; pass `organization_id`/`organization_name` or a broader "
             "`privilege_scope` for organization-wide discovery, or call "
-            "`list_cloud_organizations` to search organizations by name. If exactly "
+            "`list_organizations` to search organizations by name. If exactly "
             "one workspace is found, use it; otherwise ask the user to choose. Call "
-            "`get_default_cloud_context` to inspect your memberships.",
+            "`get_default_workspace_context` to inspect your memberships.",
             id="mcp",
         ),
         pytest.param(
diff --git a/tests/unit_tests/test_mcp_arg_trace_parity.py b/tests/unit_tests/test_mcp_arg_trace_parity.py
index 0249924b8..e40ed182d 100644
--- a/tests/unit_tests/test_mcp_arg_trace_parity.py
+++ b/tests/unit_tests/test_mcp_arg_trace_parity.py
@@ -55,7 +55,7 @@
         "streams": [{"name": SENTINEL}],
         "limit": 3,
     },
-    "list_cloud_connections": {"name_contains": SENTINEL, "limit": 5},
+    "list_connections": {"name_contains": SENTINEL, "limit": 5},
 }
 
 
diff --git a/tests/unit_tests/test_mcp_call_scope.py b/tests/unit_tests/test_mcp_call_scope.py
index 610d8bb83..a0c43d6eb 100644
--- a/tests/unit_tests/test_mcp_call_scope.py
+++ b/tests/unit_tests/test_mcp_call_scope.py
@@ -135,7 +135,7 @@ def test_organization_argument_is_recorded_without_a_workspace(
     monkeypatch: pytest.MonkeyPatch, events: list[TelemetryRecord]
 ) -> None:
     monkeypatch.setattr(CloudClient, "list_workspaces", lambda self, **_: [])
-    _call("list_cloud_workspaces", {"organization_id": ORGANIZATION})
+    _call("list_workspaces", {"organization_id": ORGANIZATION})
     assert _scope_of(events[-1]) == (None, ORGANIZATION, "arg")
 
 
@@ -158,7 +158,7 @@ def test_non_uuid_argument_is_not_recorded_or_replaced_by_the_header(
 
 
 def test_tools_without_cloud_scope_record_nulls(events: list[TelemetryRecord]) -> None:
-    _call("get_connector_info", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
     assert _scope_of(events[-1]) == (None, None, None)
 
 
@@ -167,7 +167,7 @@ def test_scope_does_not_leak_between_calls(
 ) -> None:
     _stub_workspace(monkeypatch, default=WORKSPACE)
     _call("list_custom_source_definitions", {})
-    _call("get_connector_info", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
     assert [_scope_of(event) for event in events[-2:]] == [
         (WORKSPACE, None, "default"),
         (None, None, None),
@@ -242,9 +242,9 @@ def test_resolved_organization_cannot_replace_or_mismatch_scope(
 @pytest.mark.parametrize(
     "tool",
     [
-        "describe_cloud_workspace",
-        "describe_cloud_organization",
-        "get_cloud_organization_billing_status",
+        "describe_workspace",
+        "describe_organization",
+        "get_organization_billing_status",
     ],
 )
 def test_cloud_tool_records_resolved_organization(monkeypatch, events, tool):
@@ -281,7 +281,7 @@ def test_cloud_tool_records_resolved_organization(monkeypatch, events, tool):
         "get_workspace",
         lambda **_: SimpleNamespace(workspace_id=WORKSPACE, name="Example"),
     )
-    workspace_tool = tool == "describe_cloud_workspace"
+    workspace_tool = tool == "describe_workspace"
     _call(
         tool,
         {"workspace_id": WORKSPACE}
@@ -313,7 +313,7 @@ def test_name_scoped_workspace_listing_records_only_a_single_organization(
         "get_organization",
         lambda *_, **__: SimpleNamespace(organization_name="Example"),
     )
-    _call("list_cloud_workspaces", {"organization_name": "Example"})
+    _call("list_workspaces", {"organization_name": "Example"})
     assert events[-1].success
     assert _scope_of(events[-1]) == (
         (None, ORGANIZATION, "resolved") if len(org_ids) == 1 else (None, None, None)
@@ -360,8 +360,8 @@ def get_workspace_organization_info(workspace_id: str, **_: Any) -> dict[str, st
 def test_unscoped_call_falls_back_to_the_users_default_organization(
     events: list[TelemetryRecord], verified_user: list[str]
 ) -> None:
-    _call("get_connector_info", {"connector_name": "source-faker"})
-    _call("get_connector_info", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
     assert [_scope_of(event) for event in events[-2:]] == [
         (None, ORGANIZATION, "user_default"),
         (None, ORGANIZATION, "user_default"),
@@ -398,7 +398,7 @@ def test_unscoped_call_without_a_default_workspace_records_nulls(
     monkeypatch.setattr(
         api_util, "get_user_by_auth_id", lambda *_, **__: {"userId": USER}
     )
-    _call("get_connector_info", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
     assert _scope_of(events[-1]) == (None, None, None)
     assert verified_user == []
 
@@ -409,7 +409,7 @@ def test_name_scoped_call_does_not_fall_back_to_the_users_default_organization(
     verified_user: list[str],
 ) -> None:
     monkeypatch.setattr(CloudClient, "list_workspaces", lambda self, **_: [])
-    _call("list_cloud_workspaces", {"organization_name": "Other Org"})
+    _call("list_workspaces", {"organization_name": "Other Org"})
     assert _scope_of(events[-1]) == (None, None, None)
     assert verified_user == []
 
@@ -424,8 +424,8 @@ def fail(workspace_id: str, **_: Any) -> dict[str, str]:
         raise ConnectionError(workspace_id)
 
     monkeypatch.setattr(api_util, "get_workspace_organization_info", fail)
-    _call("get_connector_info", {"connector_name": "source-faker"})
-    _call("get_connector_info", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
+    _call("get_connector_definition", {"connector_name": "source-faker"})
     assert [_scope_of(event) for event in events[-2:]] == [(None, None, None)] * 2
     assert verified_user == [OTHER_WORKSPACE]
 
@@ -448,11 +448,13 @@ def slow(workspace_id: str, **_: Any) -> dict[str, str]:
     async def run() -> None:
         async with Client(server.app) as client:
             arguments = {"connector_name": "source-faker"}
-            await asyncio.wait_for(client.call_tool("get_connector_info", arguments), 5)
+            await asyncio.wait_for(
+                client.call_tool("get_connector_definition", arguments), 5
+            )
             release.set()
             while _user_identity._pending_default_organization_lookups:
                 await asyncio.sleep(0.01)
-            await client.call_tool("get_connector_info", arguments)
+            await client.call_tool("get_connector_definition", arguments)
 
     try:
         asyncio.run(run())
diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py
index dd99ab74a..c5667dd17 100644
--- a/tests/unit_tests/test_mcp_cloud.py
+++ b/tests/unit_tests/test_mcp_cloud.py
@@ -250,19 +250,19 @@ def list_connections(
     "tool,limit_key,extra_kwargs",
     [
         pytest.param(
-            cloud_mcp.list_cloud_connectors,
+            cloud_mcp.list_connectors,
             "sources",
             {"connector_type": ConnectorType.SOURCE},
             id="sources",
         ),
         pytest.param(
-            cloud_mcp.list_cloud_connectors,
+            cloud_mcp.list_connectors,
             "destinations",
             {"connector_type": ConnectorType.DESTINATION},
             id="destinations",
         ),
         pytest.param(
-            cloud_mcp.list_cloud_connections,
+            cloud_mcp.list_connections,
             "connections",
             {"with_connection_status": False, "failing_connections_only": False},
             id="connections",
@@ -299,21 +299,21 @@ def test_mcp_cloud_list_tools_pass_limit_to_workspace(
     "tool,limit_key,forwarded_limit,extra_kwargs",
     [
         pytest.param(
-            cloud_mcp.list_cloud_connectors,
+            cloud_mcp.list_connectors,
             "sources",
             1,
             {"connector_type": ConnectorType.SOURCE},
             id="sources",
         ),
         pytest.param(
-            cloud_mcp.list_cloud_connectors,
+            cloud_mcp.list_connectors,
             "destinations",
             1,
             {"connector_type": ConnectorType.DESTINATION},
             id="destinations",
         ),
         pytest.param(
-            cloud_mcp.list_cloud_connections,
+            cloud_mcp.list_connections,
             "connections",
             None,
             {"with_connection_status": False, "failing_connections_only": False},
@@ -367,7 +367,7 @@ def test_mcp_cloud_connections_apply_limit_after_status_filter(
         lambda ctx, workspace_id=None: workspace,
     )
 
-    results = cloud_mcp.list_cloud_connections(
+    results = cloud_mcp.list_connections(
         ctx=cast(Context, object()),
         workspace_id="workspace-id",
         name_contains=None,
@@ -441,7 +441,7 @@ def test_mcp_cloud_connector_checks_map_results(
         lambda ctx, workspace_id=None: workspace,
     )
 
-    result = cloud_mcp.check_cloud_connector(
+    result = cloud_mcp.check_connector(
         ctx=cast(Context, object()),
         connector_id=connector_id,
         connector_type=explicit_type,
@@ -480,7 +480,7 @@ def test_cancel_cloud_sync_returns_cancelled_job(
 
     result = cast(
         SyncJobResult,
-        cloud_mcp.cancel_cloud_sync(
+        cloud_mcp.cancel_connection_sync(
             ctx=cast(Context, object()),
             connection_id="connection-id",
             job_id=42,
@@ -515,7 +515,7 @@ def test_cancel_cloud_sync_forwards_missing_job_id(
         lambda ctx, workspace_id=None: workspace,
     )
 
-    cloud_mcp.cancel_cloud_sync(
+    cloud_mcp.cancel_connection_sync(
         ctx=cast(Context, object()),
         connection_id="connection-id",
         job_id=None,
@@ -566,7 +566,7 @@ def get_default_context_for_user(self) -> CloudDefaultContextInfo:
 
     monkeypatch.setattr(cloud_mcp, "_get_cloud_client", lambda _: ContextClient())
 
-    result = cloud_mcp.get_default_cloud_context(cast(Context, object()))
+    result = cloud_mcp.get_default_workspace_context(cast(Context, object()))
 
     assert result.user_id == "user-id"
     assert result.member_organizations[0].organization_id == "organization-id"
@@ -614,7 +614,7 @@ def get_default_context_for_user(self) -> CloudDefaultContextInfo:
 
     monkeypatch.setattr(cloud_mcp, "_get_cloud_client", lambda _: ContextClient())
 
-    result = cloud_mcp.get_default_cloud_context(cast(Context, object()))
+    result = cloud_mcp.get_default_workspace_context(cast(Context, object()))
 
     assert result.message.startswith(
         "Default workspace ID deleted-workspace could not be verified "
@@ -643,9 +643,7 @@ def test_describe_cloud_workspace_includes_parent_organization(
         "get_workspace",
         lambda **_: SimpleNamespace(workspace_id="workspace-id", name="Workspace"),
     )
-    result = cloud_mcp.describe_cloud_workspace(
-        cast(Context, object()), workspace_id=None
-    )
+    result = cloud_mcp.describe_workspace(cast(Context, object()), workspace_id=None)
     assert result.organization_id == "org-id"
     assert result.organization_name == "Organization"
 
@@ -668,9 +666,7 @@ def test_describe_cloud_workspace_allows_missing_parent_organization(
         "get_workspace",
         lambda **_: SimpleNamespace(workspace_id="workspace-id", name="Workspace"),
     )
-    result = cloud_mcp.describe_cloud_workspace(
-        cast(Context, object()), workspace_id=None
-    )
+    result = cloud_mcp.describe_workspace(cast(Context, object()), workspace_id=None)
     assert result.organization_id is None
     assert result.organization_name is None
 
@@ -692,7 +688,7 @@ def test_get_cloud_organization_billing_status_returns_billing_info(
         "_get_cloud_client",
         lambda _: SimpleNamespace(get_organization=lambda **_: organization),
     )
-    result = cloud_mcp.get_cloud_organization_billing_status(
+    result = cloud_mcp.get_organization_billing_status(
         cast(Context, object()), organization_id=None, organization_name=None
     )
     assert result.billing_info_available is True
@@ -717,7 +713,7 @@ def get_billing_status() -> None:
         "_get_cloud_client",
         lambda _: SimpleNamespace(get_organization=lambda **_: organization),
     )
-    result = cloud_mcp.get_cloud_organization_billing_status(
+    result = cloud_mcp.get_organization_billing_status(
         cast(Context, object()), organization_id=None, organization_name=None
     )
     assert result.billing_info_available is False
@@ -741,7 +737,7 @@ def test_describe_cloud_organization_excludes_billing_fields(
         "_get_cloud_client",
         lambda _: SimpleNamespace(get_organization=lambda **_: organization),
     )
-    result = cloud_mcp.describe_cloud_organization(
+    result = cloud_mcp.describe_organization(
         cast(Context, object()), organization_id=None, organization_name=None
     )
     assert result.id == "org-id"
@@ -782,7 +778,7 @@ def set_default_workspace_for_user(
 
     monkeypatch.setattr(cloud_mcp, "_get_cloud_client", lambda _: ContextClient())
 
-    result = cloud_mcp.set_default_cloud_workspace(
+    result = cloud_mcp.set_default_workspace(
         cast(Context, object()),
         user_email="user@example.com",
         workspace_id="workspace-id",
@@ -820,7 +816,7 @@ def set_default_workspace_for_user(
     monkeypatch.setattr(cloud_mcp, "_get_cloud_client", lambda _: ContextClient())
 
     with pytest.raises(RuntimeError, match="client failed"):
-        cloud_mcp.set_default_cloud_workspace(
+        cloud_mcp.set_default_workspace(
             cast(Context, object()),
             user_email="user@example.com",
             workspace_id="workspace-id",
@@ -834,7 +830,7 @@ def set_default_workspace_for_user(
     [
         pytest.param(
             functools.partial(
-                cloud_mcp.permanently_delete_cloud_connector,
+                cloud_mcp.permanently_delete_connector,
                 connector_type=ConnectorType.SOURCE,
             ),
             "connector_id",
@@ -844,7 +840,7 @@ def set_default_workspace_for_user(
         ),
         pytest.param(
             functools.partial(
-                cloud_mcp.permanently_delete_cloud_connector,
+                cloud_mcp.permanently_delete_connector,
                 connector_type=ConnectorType.DESTINATION,
             ),
             "connector_id",
@@ -853,7 +849,7 @@ def set_default_workspace_for_user(
             id="destination",
         ),
         pytest.param(
-            cloud_mcp.permanently_delete_cloud_connection,
+            cloud_mcp.permanently_delete_connection,
             "connection_id",
             "get_connection",
             "permanently_delete_connection",
@@ -953,7 +949,7 @@ def test_permanently_delete_cloud_connector_surfaces_library_behavior(
 
     if in_use:
         with pytest.raises(AirbyteConnectorInUseError) as exc_info:
-            cloud_mcp.permanently_delete_cloud_connector(
+            cloud_mcp.permanently_delete_connector(
                 cast(Context, object()),
                 connector_id="connector-id",
                 name="delete-me connector",
@@ -963,7 +959,7 @@ def test_permanently_delete_cloud_connector_surfaces_library_behavior(
 
         assert exc_info.value is original_error
     else:
-        result = cloud_mcp.permanently_delete_cloud_connector(
+        result = cloud_mcp.permanently_delete_connector(
             cast(Context, object()),
             connector_id="connector-id",
             name="delete-me connector",
@@ -1006,7 +1002,7 @@ def test_permanently_delete_cloud_connector_respects_delete_guard_setting(
     monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_args: workspace)
     monkeypatch.setattr(cloud_mcp, "check_guid_created_in_session", lambda _: None)
 
-    result = cloud_mcp.permanently_delete_cloud_connector(
+    result = cloud_mcp.permanently_delete_connector(
         cast(Context, object()),
         connector_id="connector-id",
         name="delete-me connector",
@@ -1028,12 +1024,8 @@ def test_permanently_delete_cloud_connector_respects_delete_guard_setting(
             cloud_mcp.permanently_delete_custom_source_definition,
             id="custom-source-definition",
         ),
-        pytest.param(
-            cloud_mcp.permanently_delete_cloud_connector, id="cloud-connector"
-        ),
-        pytest.param(
-            cloud_mcp.permanently_delete_cloud_connection, id="cloud-connection"
-        ),
+        pytest.param(cloud_mcp.permanently_delete_connector, id="cloud-connector"),
+        pytest.param(cloud_mcp.permanently_delete_connection, id="cloud-connection"),
     ],
 )
 def test_delete_tools_include_shared_name_guard_tip(
@@ -1084,7 +1076,7 @@ def test_list_cloud_sync_jobs_includes_duration_and_last_updated_at(
     )
     _patch_sync_status_workspace(monkeypatch, connection)
 
-    result = cloud_mcp.list_cloud_sync_jobs(
+    result = cloud_mcp.list_connection_sync_jobs(
         ctx=cast(Context, object()),
         connection_id="connection-1",
         workspace_id=None,
@@ -1116,7 +1108,7 @@ def test_get_cloud_sync_status_includes_duration_and_last_updated_at(
     connection = SimpleNamespace(get_sync_result=MagicMock(return_value=sync_result))
     _patch_sync_status_workspace(monkeypatch, connection)
 
-    result = cloud_mcp.get_cloud_sync_status(
+    result = cloud_mcp.get_connection_sync_status(
         ctx=cast(Context, object()),
         connection_id="connection-1",
         job_id=42,
@@ -1176,7 +1168,7 @@ def test_get_cloud_sync_status_surfaces_library_errors(
     _patch_sync_status_workspace(monkeypatch, connection)
 
     with pytest.raises(AirbyteCloudError) as exc_info:
-        cloud_mcp.get_cloud_sync_status(
+        cloud_mcp.get_connection_sync_status(
             cast(Context, object()),
             connection_id="connection-1",
             job_id=42,
@@ -1201,7 +1193,7 @@ def test_get_cloud_sync_logs_without_jobs_explains_no_syncs_yet(
     _patch_sync_status_workspace(monkeypatch, connection)
 
     with pytest.raises(AirbyteMissingResourceError) as exc_info:
-        cloud_mcp.get_cloud_sync_logs(
+        cloud_mcp.get_connection_sync_logs(
             cast(Context, object()),
             connection_id="connection-1",
             job_id=None,
@@ -1212,10 +1204,10 @@ def test_get_cloud_sync_logs_without_jobs_explains_no_syncs_yet(
         )
 
     assert "Connection 'connection-1' has no sync jobs yet." in str(exc_info.value)
-    assert "run_cloud_sync" in (exc_info.value.guidance or "")
+    assert "run_connection_sync" in (exc_info.value.guidance or "")
 
     with pytest.raises(AirbyteMissingResourceError) as explicit_exc_info:
-        cloud_mcp.get_cloud_sync_logs(
+        cloud_mcp.get_connection_sync_logs(
             cast(Context, object()),
             connection_id="connection-1",
             job_id=42,
@@ -1246,7 +1238,7 @@ def test_set_cloud_connection_selected_streams_surfaces_library_error(
     monkeypatch.setattr(cloud_mcp, "check_guid_created_in_session", lambda _: None)
 
     with pytest.raises(AirbyteLibInputError) as exc_info:
-        cloud_mcp.set_cloud_connection_selected_streams(
+        cloud_mcp.set_connection_selected_streams(
             cast(Context, object()),
             connection_id="connection-1",
             stream_names=["missing", "orders"],
@@ -1286,7 +1278,7 @@ def test_deploy_connector_to_cloud_routes_by_type(
     registry_getter: str,
     explicit_type: bool,
 ) -> None:
-    """Verify `deploy_connector_to_cloud` dispatches to the matching getter and deploy path."""
+    """Verify `deploy_connector` dispatches to the matching getter and deploy path."""
     connector_cls = Source if connector_type == ConnectorType.SOURCE else Destination
     connector = MagicMock(spec=connector_cls)
     connector.config_spec = {"type": "object"}
@@ -1318,7 +1310,7 @@ def fake_deploy(**kwargs: object) -> SimpleNamespace:
     monkeypatch.setattr(
         cloud_mcp, "register_guid_created_in_session", registered.append
     )
-    result = cloud_mcp.deploy_connector_to_cloud(
+    result = cloud_mcp.deploy_connector(
         cast(Context, object()),
         name="My Connector",
         connector_name=connector_name,
@@ -1343,7 +1335,7 @@ def fake_deploy(**kwargs: object) -> SimpleNamespace:
 def test_deploy_connector_to_cloud_rejects_unknown_prefix() -> None:
     """Verify a non-canonical name without an explicit type raises `AirbyteLibInputError`."""
     with pytest.raises(AirbyteLibInputError, match="Cannot infer connector type"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             cast(Context, object()),
             name="My Connector",
             connector_name="faker",
@@ -1389,7 +1381,7 @@ def fake_resolve_config(**kwargs: object) -> dict[str, Any]:
         cloud_mcp, "register_guid_created_in_session", registered.append
     )
 
-    result = cloud_mcp.deploy_connector_to_cloud(
+    result = cloud_mcp.deploy_connector(
         cast(Context, object()),
         name="My custom source",
         connector_name=None,
@@ -1437,7 +1429,7 @@ def test_deploy_connector_to_cloud_requires_one_definition_selector(
 ) -> None:
     """Exactly one of a registry connector name or custom definition ID is required."""
     with pytest.raises(AirbyteLibInputError, match="exactly one"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             cast(Context, object()),
             name="My connector",
             connector_name=connector_name,
@@ -1453,7 +1445,7 @@ def test_deploy_connector_to_cloud_requires_one_definition_selector(
 def test_deploy_connector_to_cloud_rejects_custom_destination() -> None:
     """Custom source definitions cannot be deployed as destinations."""
     with pytest.raises(AirbyteLibInputError, match="only be deployed as sources"):
-        cloud_mcp.deploy_connector_to_cloud(
+        cloud_mcp.deploy_connector(
             cast(Context, object()),
             name="My connector",
             connector_name=None,
@@ -1491,7 +1483,7 @@ def fake_deploy(**kwargs: object) -> SimpleNamespace:
         cloud_mcp, "register_guid_created_in_session", registered.append
     )
 
-    result = cloud_mcp.deploy_connector_to_cloud(
+    result = cloud_mcp.deploy_connector(
         cast(Context, object()),
         name="My custom source",
         connector_name=None,
@@ -1592,7 +1584,7 @@ def fail_lookup(_definition_id: str) -> None:
         cloud_mcp, "get_connector_metadata_by_definition_id", fail_lookup
     )
 
-    result = cloud_mcp.create_connection_on_cloud(
+    result = cloud_mcp.create_connection(
         cast(Context, object()),
         connection_name="My Connection",
         source_id="source-id",
@@ -1628,7 +1620,7 @@ def test_create_connection_on_cloud_defaults_to_suggested_streams(
         lambda definition_id: metadata if definition_id == "definition-id" else None,
     )
 
-    result = cloud_mcp.create_connection_on_cloud(
+    result = cloud_mcp.create_connection(
         cast(Context, object()),
         connection_name="My Connection",
         source_id="source-id",
@@ -1664,7 +1656,7 @@ def test_create_connection_on_cloud_without_suggested_streams_raises(
     )
 
     with pytest.raises(AirbyteLibInputError, match="suggested streams"):
-        cloud_mcp.create_connection_on_cloud(
+        cloud_mcp.create_connection(
             cast(Context, object()),
             connection_name="My Connection",
             source_id="source-id",
@@ -1686,7 +1678,7 @@ def test_create_connection_on_cloud_unknown_destination_names_arg(
     )
 
     with pytest.raises(AirbyteMissingResourceError, match="`destination_id`"):
-        cloud_mcp.create_connection_on_cloud(
+        cloud_mcp.create_connection(
             cast(Context, object()),
             connection_name="My Connection",
             source_id="source-id",
@@ -1708,7 +1700,7 @@ def test_create_connection_on_cloud_unknown_source_names_arg(
     )
 
     with pytest.raises(AirbyteMissingResourceError, match="`source_id`"):
-        cloud_mcp.create_connection_on_cloud(
+        cloud_mcp.create_connection(
             cast(Context, object()),
             connection_name="My Connection",
             source_id="source-id",
@@ -1787,7 +1779,7 @@ def test_list_cloud_connectors_returns_both_kinds(
     """The combined tool maps each connector's type and enabled features."""
     _patch_combined_listing(monkeypatch)
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -1801,7 +1793,7 @@ def test_list_cloud_connectors_returns_both_kinds(
     assert results[0].enabled_features == cloud_mcp.FEATURES_NOT_CHECKED
     assert results[1].enabled_features == cloud_mcp.FEATURES_NOT_CHECKED
 
-    resolved = cloud_mcp.list_cloud_connectors(
+    resolved = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -1824,7 +1816,7 @@ def test_list_cloud_connectors_filters(
     """`connector_type` and `feature_filter` narrow the combined listing."""
     workspace = _patch_combined_listing(monkeypatch)
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         connector_type=ConnectorType.DESTINATION,
@@ -2442,7 +2434,7 @@ def test_get_cloud_search_status_filters_by_stream_name(
     connector = _RecordingExecuteConnector()
     workspace = _execute_workspace(monkeypatch, connector)
 
-    result = cloud_mcp.get_cloud_search_status(
+    result = cloud_mcp.get_search_index_status(
         None,
         connector_id="destination-1",
         stream_name=stream_name,
@@ -2493,7 +2485,7 @@ def test_get_cloud_search_status_namespace_filter_and_miss_warning(
     connector = _RecordingExecuteConnector()
     _execute_workspace(monkeypatch, connector)
 
-    result = cloud_mcp.get_cloud_search_status(
+    result = cloud_mcp.get_search_index_status(
         None,
         connector_id="destination-1",
         stream_name=stream_name,
@@ -2523,7 +2515,7 @@ def test_get_cloud_search_status_miss_warning_caps_stream_list(
     monkeypatch.setattr(connector, "get_search_status", lambda: status)
     _execute_workspace(monkeypatch, connector)
 
-    result = cloud_mcp.get_cloud_search_status(
+    result = cloud_mcp.get_search_index_status(
         None,
         connector_id="source-1",
         stream_name="missing",
@@ -2570,7 +2562,7 @@ def test_search_tools_use_connector_type_without_probe(
         prompt="refund requests",
         workspace_id=None,
     )
-    cloud_mcp.get_cloud_search_status(
+    cloud_mcp.get_search_index_status(
         None,
         connector_id="connector-1",
         connector_type=connector_type,
@@ -2644,7 +2636,7 @@ def test_search_tools_return_search_disabled_errors(
         prompt="refund requests",
         workspace_id=None,
     )
-    status = cloud_mcp.get_cloud_search_status(
+    status = cloud_mcp.get_search_index_status(
         None,
         connector_id="source-1",
         stream_name="issues",
@@ -2655,8 +2647,8 @@ def test_search_tools_return_search_disabled_errors(
         expected_message,
         "Once search is enabled, the connector's synced data must be indexed before "
         "searches return results. New syncs are indexed as they run; data already in the "
-        "destination is indexed only after `request_cloud_search_index_backfill`. Call "
-        "`get_cloud_search_status` to check which streams are indexed and whether their "
+        "destination is indexed only after `request_search_index_backfill`. Call "
+        "`get_search_index_status` to check which streams are indexed and whether their "
         "backfill is complete.",
     ]
     assert search.errors == expected_errors
@@ -2693,7 +2685,7 @@ def test_search_disabled_errors_are_redacted(monkeypatch: pytest.MonkeyPatch) ->
 
 
 @pytest.mark.parametrize(
-    "tool_name", ["execute_external_search_query", "get_cloud_search_status"]
+    "tool_name", ["execute_external_search_query", "get_search_index_status"]
 )
 def test_search_tools_reraise_without_enablement_url(
     monkeypatch: pytest.MonkeyPatch,
@@ -2737,7 +2729,7 @@ def test_search_tools_are_always_advertised(
     else:
         monkeypatch.setenv(MCP_INSIDERS_ENV_VAR, insiders_env)
     names = {tool.name for tool in asyncio.run(server.app.list_tools())}
-    assert {"execute_external_search_query", "get_cloud_search_status"} <= names
+    assert {"execute_external_search_query", "get_search_index_status"} <= names
 
 
 @pytest.mark.parametrize(
@@ -2784,7 +2776,7 @@ def _describe_details() -> CloudConnectorDetailsResult:
 def test_describe_cloud_connector_forwards_id_and_toggles(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """`describe_cloud_connector` forwards the ID and all `with_*` toggles."""
+    """`describe_connector` forwards the ID and all `with_*` toggles."""
     details = _describe_details()
     describe = MagicMock(return_value=details)
     monkeypatch.setattr(cloud_mcp, "_describe_cloud_connector", describe)
@@ -2795,7 +2787,7 @@ def test_describe_cloud_connector_forwards_id_and_toggles(
         cloud_mcp, "_get_cloud_workspace", lambda *args, **kwargs: workspace
     )
 
-    result = cloud_mcp.describe_cloud_connector(
+    result = cloud_mcp.describe_connector(
         ctx=cast(Context, object()),
         connector_id="connector-1",
         workspace_id="workspace-1",
@@ -2849,7 +2841,7 @@ def test_describe_cloud_connector_explicit_type_skips_lookup(
         cloud_mcp, "_get_cloud_workspace", lambda *args, **kwargs: workspace
     )
 
-    result = cloud_mcp.describe_cloud_connector(
+    result = cloud_mcp.describe_connector(
         ctx=cast(Context, object()),
         connector_id="connector-1",
         connector_type=connector_type,
@@ -3097,7 +3089,7 @@ def test_describe_cloud_connection_includes_status_and_schedule(
     workspace = SimpleNamespace(get_connection=lambda **_: connection)
     monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_args: workspace)
 
-    result = cloud_mcp.describe_cloud_connection(
+    result = cloud_mcp.describe_connection(
         cast(Context, object()),
         connection_id="connection-1",
         workspace_id=None,
@@ -3316,7 +3308,7 @@ def test_list_cloud_connectors_feature_filter_probe_failure_marks_unknown(
         ],
     )
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -3361,7 +3353,7 @@ def test_list_cloud_connectors_feature_filter_connector_failure_keeps_probing(
         ],
     )
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -3381,7 +3373,7 @@ def test_list_cloud_connectors_limit_must_be_positive(
 ) -> None:
     """`limit=0` raises `AirbyteLibInputError` before listing connectors."""
     with pytest.raises(AirbyteLibInputError, match="`limit` must be greater than 0"):
-        cloud_mcp.list_cloud_connectors(
+        cloud_mcp.list_connectors(
             None,
             workspace_id=None,
             name_contains=None,
@@ -3411,7 +3403,7 @@ def test_list_cloud_connectors_feature_filter_not_enabled_still_excluded(
         ],
     )
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -3440,7 +3432,7 @@ def test_list_cloud_connectors_feature_filter_limit_counts_unknown(
         ],
     )
 
-    results = cloud_mcp.list_cloud_connectors(
+    results = cloud_mcp.list_connectors(
         None,
         workspace_id=None,
         name_contains=None,
@@ -3584,7 +3576,7 @@ def fail(*_args: object, **_kwargs: object) -> None:
     @server.tool
     def caught_error_result() -> dict[str, Any]:
         if result_path == "billing":
-            return cloud_mcp.get_cloud_organization_billing_status(
+            return cloud_mcp.get_organization_billing_status(
                 cast(Context, object()), organization_id=None, organization_name=None
             ).model_dump()
         if result_path == "feature-warning":
@@ -3673,7 +3665,7 @@ def test_request_cloud_search_index_backfill_forwards_args(
 ) -> None:
     workspace = _backfill_workspace(monkeypatch)
 
-    cloud_mcp.request_cloud_search_index_backfill(
+    cloud_mcp.request_search_index_backfill(
         None,
         connection_id="connection-1",
         streams=streams,
@@ -3695,7 +3687,7 @@ def test_request_cloud_search_index_backfill_rejects_invalid_start_time(
     workspace = _backfill_workspace(monkeypatch)
 
     with pytest.raises(AirbyteLibInputError, match="ISO 8601"):
-        cloud_mcp.request_cloud_search_index_backfill(
+        cloud_mcp.request_search_index_backfill(
             None,
             connection_id="connection-1",
             backfill_start_time="last tuesday",
@@ -3708,15 +3700,13 @@ def test_request_cloud_search_index_backfill_rejects_invalid_start_time(
 def test_request_cloud_search_index_backfill_is_a_write_tool() -> None:
     """Requesting a backfill queues index work, so the tool is a closed-world write."""
     (annotations,) = [
-        a
-        for f, a in _REGISTERED_TOOLS
-        if f is cloud_mcp.request_cloud_search_index_backfill
+        a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.request_search_index_backfill
     ]
     assert annotations["readOnlyHint"] is False
     assert annotations["destructiveHint"] is False
     assert annotations["idempotentHint"] is False
     assert annotations["openWorldHint"] is False
-    policy = _TOOL_POLICIES["request_cloud_search_index_backfill"]
+    policy = _TOOL_POLICIES["request_search_index_backfill"]
     assert policy.external_access is True
     assert policy.pipeline_change is False
 
@@ -3894,7 +3884,7 @@ def fake_deploy_destination(**kwargs: object) -> SimpleNamespace:
     monkeypatch.setattr(cloud_mcp, "get_noop_destination", object)
     monkeypatch.setattr(cloud_mcp, "register_guid_created_in_session", lambda _: None)
 
-    result = cloud_mcp.deploy_noop_destination_to_cloud(
+    result = cloud_mcp.deploy_noop_destination(
         cast(Context, object()),
         destination_name="my-noop",
         workspace_id=None,
@@ -3905,7 +3895,7 @@ def fake_deploy_destination(**kwargs: object) -> SimpleNamespace:
     assert deployed["unique"] is False
     assert "noop-id" in result
 
-    default_result = cloud_mcp.deploy_noop_destination_to_cloud(
+    default_result = cloud_mcp.deploy_noop_destination(
         cast(Context, object()), workspace_id=None
     )
     assert deployed["name"] == "No-op Destination"
@@ -3915,7 +3905,7 @@ def fake_deploy_destination(**kwargs: object) -> SimpleNamespace:
 
 def test_deploy_noop_destination_to_cloud_params_have_descriptions() -> None:
     """Verify every No-op deploy tool parameter exposes a schema description."""
-    signature = inspect.signature(cloud_mcp.deploy_noop_destination_to_cloud)
+    signature = inspect.signature(cloud_mcp.deploy_noop_destination)
     for param_name in ("destination_name", "workspace_id", "unique"):
         annotation = signature.parameters[param_name].annotation
         field_infos = [
diff --git a/tests/unit_tests/test_mcp_connection_schedules.py b/tests/unit_tests/test_mcp_connection_schedules.py
index c8f0d2bfa..a42d6750c 100644
--- a/tests/unit_tests/test_mcp_connection_schedules.py
+++ b/tests/unit_tests/test_mcp_connection_schedules.py
@@ -96,7 +96,7 @@ async def _update(server: FastMCP, **arguments: object) -> str:
     """Call the registered tool through the serialized MCP boundary."""
     async with Client(server) as client:
         result = await client.call_tool(
-            "update_cloud_connection",
+            "update_connection",
             {"connection_id": "12345678-1234-1234-1234-1234567890ab", **arguments},
         )
     return str(result)
@@ -110,7 +110,7 @@ def test_interval_schema() -> None:
     async def inspect_tools() -> None:
         async with Client(server) as client:
             tools = {tool.name: tool for tool in await client.list_tools()}
-        update = tools["update_cloud_connection"]
+        update = tools["update_connection"]
         interval = update.input_schema["properties"]["interval_hours"]
         integer_schema = next(
             option for option in interval["anyOf"] if option["type"] == "integer"
@@ -133,12 +133,12 @@ def test_create_manual_default_description() -> None:
     async def inspect_tools() -> None:
         async with Client(server) as client:
             tools = {tool.name: tool for tool in await client.list_tools()}
-        create = tools["create_connection_on_cloud"]
+        create = tools["create_connection"]
         description = (create.description or "").lower()
         assert "manual" in description
         assert "default" in description
         assert "automatic syncs" in description
-        assert "update_cloud_connection" in description
+        assert "update_connection" in description
 
     asyncio.run(inspect_tools())
 
@@ -152,7 +152,7 @@ def test_direct_update_rejects_invalid_intervals_before_status_change(
     _server, http, lookup, patch = schedule_backend
 
     with pytest.raises(AirbyteLibInputError, match="positive whole number"):
-        cloud_mcp.update_cloud_connection(
+        cloud_mcp.update_connection(
             ctx=None,
             connection_id="12345678-1234-1234-1234-1234567890ab",
             enabled=True,
@@ -300,7 +300,7 @@ def test_update_remains_hidden_in_readonly_mode(
     server = FastMCP("readonly-schedule-tests")
     cloud_mcp.register_cloud_tools(server)
     monkeypatch.setenv(MCP_READONLY_MODE_ENV_VAR, "1")
-    tool = asyncio.run(server.get_tool("update_cloud_connection"))
+    tool = asyncio.run(server.get_tool("update_connection"))
 
     assert not _tool_utils.airbyte_readonly_mode_filter(tool.to_mcp_tool(), server)
 
diff --git a/tests/unit_tests/test_mcp_connector_registry.py b/tests/unit_tests/test_mcp_connector_registry.py
index 4ffc0977e..39dede3f2 100644
--- a/tests/unit_tests/test_mcp_connector_registry.py
+++ b/tests/unit_tests/test_mcp_connector_registry.py
@@ -11,13 +11,16 @@
 
 from airbyte import exceptions as exc
 from airbyte.mcp._tool_utils import _mcp_module_for_tool
-from airbyte.mcp.interactive import show_connectors_list, show_workspace_sync_status
+from airbyte.mcp.interactive import (
+    show_connector_definitions,
+    show_workspace_sync_status,
+)
 from airbyte.mcp.interactive._registry_ui import (
     _connector_metadata_to_public_summary,
     _list_public_registry_connectors,
 )
 from airbyte.mcp.interactive._shared_models import ConnectorType, SupportLevel
-from airbyte.mcp.guidance import get_api_docs_urls, get_connector_info
+from airbyte.mcp.guidance import get_api_docs_urls, get_connector_definition
 from airbyte.mcp.registry import get_available_connectors
 from airbyte.registry import (
     ApiDocsUrl,
@@ -263,12 +266,12 @@ def test_show_connectors_list_limits_text_and_meta_payload() -> None:
 
     with patch(
         "airbyte.mcp.interactive._registry_ui._list_public_registry_connectors"
-    ) as mock_list_connectors:
-        mock_list_connectors.return_value = [
+    ) as mock_list_connector_definitions:
+        mock_list_connector_definitions.return_value = [
             _connector_metadata_to_public_summary(connector)
             for connector in connector_metadata
         ]
-        result = show_connectors_list()
+        result = show_connector_definitions()
 
     text_content = result.content[0]
     assert isinstance(text_content, TextContent)
@@ -289,9 +292,9 @@ def test_list_public_registry_connectors_uses_existing_registry_tool() -> None:
     """Test that the interactive list starts from the existing MCP registry tool."""
     with (
         patch(
-            "airbyte.mcp.interactive._registry_ui._list_connectors",
+            "airbyte.mcp.interactive._registry_ui._list_connector_definitions",
             return_value=[],
-        ) as mock_list_connectors,
+        ) as mock_list_connector_definitions,
         patch("airbyte.mcp.interactive._registry_ui.get_connector_metadata"),
     ):
         _list_public_registry_connectors(
@@ -302,7 +305,7 @@ def test_list_public_registry_connectors_uses_existing_registry_tool() -> None:
             limit=10,
         )
 
-    mock_list_connectors.assert_called_once_with(
+    mock_list_connector_definitions.assert_called_once_with(
         keyword_filter=None,
         connector_type_filter="source",
         install_types=None,
@@ -378,7 +381,7 @@ def test_list_public_registry_connectors_applies_filters(
     """Test that connector metadata filters are applied by the interactive list."""
     with (
         patch(
-            "airbyte.mcp.interactive._registry_ui._list_connectors",
+            "airbyte.mcp.interactive._registry_ui._list_connector_definitions",
             return_value=list(connector_metadata_cache),
         ),
         patch(
@@ -404,7 +407,7 @@ def test_show_connectors_list_rejects_negative_limit() -> None:
     with pytest.raises(
         exc.AirbyteLibInputError, match="Limit parameter must be non-negative."
     ):
-        show_connectors_list(limit=-1)
+        show_connector_definitions(limit=-1)
 
 
 def test_show_connectors_list_uses_resolved_registry_url(
@@ -414,7 +417,7 @@ def test_show_connectors_list_uses_resolved_registry_url(
     registry_url = "file:///tmp/local_registry.json"
     with (
         patch(
-            "airbyte.mcp.interactive._registry_ui._list_connectors",
+            "airbyte.mcp.interactive._registry_ui._list_connector_definitions",
             return_value=list(connector_metadata_cache),
         ),
         patch(
@@ -426,7 +429,7 @@ def test_show_connectors_list_uses_resolved_registry_url(
             return_value=registry_url,
         ),
     ):
-        result = show_connectors_list(limit=1)
+        result = show_connector_definitions(limit=1)
 
     assert result.meta is not None
     assert result.meta["airbyte_mcp_raw_result"]["registry_url"] == registry_url
@@ -437,7 +440,7 @@ def test_show_connectors_list_rejects_conflicting_support_filters() -> None:
     with pytest.raises(
         ValueError, match="Cannot specify both `certified` and `support_level`"
     ):
-        show_connectors_list(certified=True, support_level="community")
+        show_connector_definitions(certified=True, support_level="community")
 
 
 @pytest.mark.parametrize(
@@ -468,7 +471,7 @@ def test_interactive_tools_are_filtered_by_ui_support(
         tools = asyncio.run(app.list_tools())
 
     tool_names = {tool.name for tool in tools}
-    assert ("show_connectors_list" in tool_names) is expected_visible
+    assert ("show_connector_definitions" in tool_names) is expected_visible
     assert ("show_workspace_sync_status" in tool_names) is expected_visible
 
 
@@ -485,7 +488,7 @@ def test_interactive_tools_are_rejected_by_tool_filter_without_ui_support() -> N
     interactive.register_interactive_tools(app)
 
     with pytest.raises(ValueError, match="not available"):
-        asyncio.run(app.call_tool("show_connectors_list"))
+        asyncio.run(app.call_tool("show_connector_definitions"))
 
 
 def test_interactive_tools_include_prefab_metadata() -> None:
@@ -498,7 +501,7 @@ def test_interactive_tools_include_prefab_metadata() -> None:
     interactive.register_interactive_tools(app)
 
     provider = getattr(app, "_local_provider")
-    tool = provider._components["tool:show_connectors_list@"]  # noqa: SLF001
+    tool = provider._components["tool:show_connector_definitions@"]  # noqa: SLF001
 
     assert tool.meta is not None
     assert tool.meta["ui"]["resourceUri"] == "ui://prefab/renderer.html"
@@ -506,13 +509,13 @@ def test_interactive_tools_include_prefab_metadata() -> None:
 
 def test_mcp_module_for_tool_uses_nearest_public_module() -> None:
     """Test that tools in private implementation modules use their public module."""
-    assert _mcp_module_for_tool(show_connectors_list) == "interactive"
+    assert _mcp_module_for_tool(show_connector_definitions) == "interactive"
     assert _mcp_module_for_tool(show_workspace_sync_status) == "interactive"
     assert _mcp_module_for_tool(get_api_docs_urls) == "guidance"
 
 
 def test_get_connector_info_resolves_spec_from_registry_without_docker() -> None:
-    """`get_connector_info` must resolve the config spec over HTTP, never via install.
+    """`get_connector_definition` must resolve the config spec over HTTP, never via install.
 
     The spec is fetched from the public connector registry keyed by version, so it
     works in a hosted, no-Docker runtime without the unbounded `connector.install()`
@@ -536,7 +539,7 @@ def test_get_connector_info_resolves_spec_from_registry_without_docker() -> None
             return_value=cloud_spec,
         ) as mock_get_spec,
     ):
-        result = get_connector_info("source-faker")
+        result = get_connector_definition("source-faker")
 
     assert not isinstance(result, str)
     assert result.config_spec_jsonschema == cloud_spec
@@ -585,14 +588,14 @@ def test_get_connector_info_finds_java_connector_without_docker() -> None:
         ),
     ):
         assert get_available_connectors() == []
-        result = get_connector_info("source-postgres")
+        result = get_connector_definition("source-postgres")
 
     assert not isinstance(result, str)
     assert result.connector_name == "source-postgres"
 
 
 def test_get_connector_info_falls_back_to_oss_spec() -> None:
-    """When the `cloud` spec is unavailable, `get_connector_info` uses the `oss` spec."""
+    """When the `cloud` spec is unavailable, `get_connector_definition` uses the `oss` spec."""
     connector = MagicMock()
     connector.name = "source-faker"
     connector.docs_url = "https://docs.airbyte.com/integrations/sources/faker"
@@ -611,7 +614,7 @@ def test_get_connector_info_falls_back_to_oss_spec() -> None:
             side_effect=[None, oss_spec],
         ) as mock_get_spec,
     ):
-        result = get_connector_info("source-faker")
+        result = get_connector_definition("source-faker")
 
     assert not isinstance(result, str)
     assert result.config_spec_jsonschema == oss_spec
@@ -640,7 +643,7 @@ def test_get_connector_info_spec_none_when_registry_has_no_spec() -> None:
             return_value=None,
         ),
     ):
-        result = get_connector_info("source-faker")
+        result = get_connector_definition("source-faker")
 
     assert not isinstance(result, str)
     assert result.config_spec_jsonschema is None
diff --git a/tests/unit_tests/test_mcp_datadog.py b/tests/unit_tests/test_mcp_datadog.py
index 98baa06b0..5de70e157 100644
--- a/tests/unit_tests/test_mcp_datadog.py
+++ b/tests/unit_tests/test_mcp_datadog.py
@@ -1221,7 +1221,7 @@ def execute(
         return ToolResult(content=private_payload, is_error=mode == "error")
 
     @app.tool()
-    def run_sql_query() -> str:
+    def run_local_sql_query() -> str:
         return "private synthetic result"
 
     @app.tool()
@@ -1327,7 +1327,7 @@ async def run():
                         )
                         for mode in ("ok", "error", "raise")
                     ],
-                    ("tools/call", {"name": "run_sql_query", "arguments": {}}),
+                    ("tools/call", {"name": "run_local_sql_query", "arguments": {}}),
                     ("tools/call", {"name": "unknown_tool", "arguments": {}}),
                     (
                         "tools/call",
@@ -1913,13 +1913,13 @@ async def dispatch(_ctx):
             "execute_external_api_query",
             "execute_external_api_query",
             "execute_external_api_query",
-            "run_sql_query",
+            "run_local_sql_query",
             "unknown_tool",
         ]
         assert [span.resource for span in primary] == [
             "unknown_tool",
             *(["execute_external_api_query"] * 3),
-            "run_sql_query",
+            "run_local_sql_query",
             "unknown_tool",
         ]
         for index in (1, 2, 3):
diff --git a/tests/unit_tests/test_mcp_http.py b/tests/unit_tests/test_mcp_http.py
index 2dbadab57..8d5c261a8 100644
--- a/tests/unit_tests/test_mcp_http.py
+++ b/tests/unit_tests/test_mcp_http.py
@@ -29,7 +29,7 @@
 
 UI_EXTENSION = {"io.modelcontextprotocol/ui": {}}
 UI_TOOL_NAMES = {
-    "show_connectors_list",
+    "show_connector_definitions",
     "show_workspace_sync_status",
     "show_connection_sync_history",
 }
diff --git a/tests/unit_tests/test_mcp_http_main.py b/tests/unit_tests/test_mcp_http_main.py
index 62eb5c489..fe99f7255 100644
--- a/tests/unit_tests/test_mcp_http_main.py
+++ b/tests/unit_tests/test_mcp_http_main.py
@@ -213,10 +213,10 @@ def test_main_sets_hosted_instructions(monkeypatch: pytest.MonkeyPatch) -> None:
         "request headers",
     ):
         assert local_only not in instructions
-    assert "get_default_cloud_context" in instructions
+    assert "get_default_workspace_context" in instructions
     assert "Deletion guard" in instructions
     assert "Read-only mode" in instructions
-    assert "list_cloud_organizations" in instructions
+    assert "list_organizations" in instructions
 
 
 def test_stdio_instructions_keep_local_guidance() -> None:
@@ -226,4 +226,4 @@ def test_stdio_instructions_keep_local_guidance() -> None:
     assert "AIRBYTE_CLOUD_BEARER_TOKEN" in instructions
     assert "AIRBYTE_PROJECT_DIR" in instructions
     assert "Local connector execution" in instructions
-    assert "list_cloud_organizations" in instructions
+    assert "list_organizations" in instructions
diff --git a/tests/unit_tests/test_mcp_local_tool_descriptions.py b/tests/unit_tests/test_mcp_local_tool_descriptions.py
index 2ac6269c7..c1fd2ad10 100644
--- a/tests/unit_tests/test_mcp_local_tool_descriptions.py
+++ b/tests/unit_tests/test_mcp_local_tool_descriptions.py
@@ -12,18 +12,18 @@
 _CONFIG_HELP_TEXT = local._CONFIG_HELP.strip()
 
 _TOOLS_WITH_CONFIG_HELP = (
-    "validate_connector_config",
-    "list_source_streams",
-    "get_source_stream_json_schema",
-    "read_source_stream_records",
-    "get_stream_previews",
-    "sync_source_to_cache",
+    "validate_local_connector_config",
+    "list_local_source_streams",
+    "get_local_source_stream_json_schema",
+    "read_local_source_stream_records",
+    "get_local_stream_previews",
+    "sync_local_source_to_cache",
 )
 _TOOLS_WITHOUT_CONFIG_HELP = (
-    "list_dotenv_secrets",
-    "list_cached_streams",
-    "describe_default_cache",
-    "run_sql_query",
+    "list_local_dotenv_secrets",
+    "list_local_cached_streams",
+    "describe_local_default_cache",
+    "run_local_sql_query",
 )
 
 
diff --git a/tests/unit_tests/test_mcp_policy_middleware.py b/tests/unit_tests/test_mcp_policy_middleware.py
index b457eee61..df96abde6 100644
--- a/tests/unit_tests/test_mcp_policy_middleware.py
+++ b/tests/unit_tests/test_mcp_policy_middleware.py
@@ -326,7 +326,7 @@ async def call(tool_name: str, annotations: SimpleNamespace) -> str:
         )
         return await PolicyGuardMiddleware().on_call_tool(context, call_next)
 
-    _MW_TOOL_MODULES["destination_smoke_test"] = "local"
+    _MW_TOOL_MODULES["run_local_destination_smoke_test"] = "local"
     _MW_TOOL_MODULES["sync_connector"] = "local"
     # No `ToolPolicy` registration: the tools exercise the real default
     # (non-read-only -> pipeline_change=True).
@@ -336,7 +336,7 @@ async def call(tool_name: str, annotations: SimpleNamespace) -> str:
     assert (
         asyncio.run(
             call(
-                "destination_smoke_test",
+                "run_local_destination_smoke_test",
                 SimpleNamespace(destructive_hint=True),
             )
         )
diff --git a/tests/unit_tests/test_mcp_tool_annotations.py b/tests/unit_tests/test_mcp_tool_annotations.py
index 107d62e3e..3cede87d7 100644
--- a/tests/unit_tests/test_mcp_tool_annotations.py
+++ b/tests/unit_tests/test_mcp_tool_annotations.py
@@ -5,6 +5,7 @@
 
 import airbyte.mcp.server  # noqa: F401  # Importing registers every MCP tool module.
 from airbyte.mcp import cloud as cloud_mcp
+from airbyte.mcp.registry import list_connector_definitions
 from fastmcp_extensions.decorators import _REGISTERED_TOOLS  # noqa: PLC2701
 
 READ_ONLY_NAME_PREFIXES = ("list_", "describe_", "get_", "check_")
@@ -19,11 +20,31 @@ def _cloud_tool_entries() -> list[tuple[object, dict]]:
     return [(f, a) for f, a in _REGISTERED_TOOLS if f.__module__ == cloud_mcp.__name__]
 
 
+def test_tool_names_are_unique_and_cloud_prefix_free() -> None:
+    """Tool registrations expose only the new unambiguous names."""
+    names = [f.__name__ for f, _ in _REGISTERED_TOOLS]
+    assert len(names) == len(set(names))
+    assert not any("cloud" in name for name in names)
+    assert "list_connector_definitions" in names
+    assert "list_connectors" in names
+    assert "get_connector_definition" in names
+    assert "show_connector_definitions" in names
+
+
+def test_connector_listing_descriptions_distinguish_registry_from_workspace() -> None:
+    """Registry types and deployed workspace connectors have separate discovery tools."""
+    registry_description = list_connector_definitions.__doc__ or ""
+    assert "connector types available to deploy" in registry_description
+    assert "`list_connectors`" in registry_description
+    assert "`list_custom_source_definitions`" in registry_description
+    workspace_description = cloud_mcp.list_connectors.__doc__ or ""
+    assert "deployed source and destination" in workspace_description
+    assert "`list_connector_definitions`" in workspace_description
+
+
 def test_check_cloud_connector_is_read_only_and_idempotent() -> None:
-    """`check_cloud_connector` only returns the connector check result."""
-    (annotations,) = [
-        a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.check_cloud_connector
-    ]
+    """`check_connector` only returns the connector check result."""
+    (annotations,) = [a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.check_connector]
     assert annotations["readOnlyHint"] is True
     assert annotations["idempotentHint"] is True
 
diff --git a/tests/unit_tests/test_mcp_tool_filters.py b/tests/unit_tests/test_mcp_tool_filters.py
index 9dd7ae66f..d6b8a43e9 100644
--- a/tests/unit_tests/test_mcp_tool_filters.py
+++ b/tests/unit_tests/test_mcp_tool_filters.py
@@ -149,7 +149,7 @@ def test_ui_tool_wire_meta_carries_only_standard_ui_key() -> None:
 
     from airbyte.mcp.server import app as airbyte_app
 
-    tool = asyncio.run(airbyte_app.get_tool("show_connectors_list"))
+    tool = asyncio.run(airbyte_app.get_tool("show_connector_definitions"))
     wire = tool.to_mcp_tool().model_dump(by_alias=True)
 
     assert (wire.get("_meta") or {}).get("ui", {}).get("resourceUri")
@@ -212,7 +212,7 @@ def test_sync_tools_stay_visible_when_pipeline_changes_are_disabled(
 ) -> None:
     assert not mcp_config
     monkeypatch.setenv(MCP_ALLOW_PIPELINE_CHANGES_ENV_VAR, "0")
-    for tool_name in ("run_cloud_sync", "cancel_cloud_sync"):
+    for tool_name in ("run_connection_sync", "cancel_connection_sync"):
         tool = _policy_tool(tool_name, pipeline_change=False)
         assert _tool_utils.airbyte_readonly_mode_filter(tool, APP)
 
@@ -224,11 +224,11 @@ def test_non_read_only_tools_hidden_in_readonly_mode(
     """Read-only mode hides non-read-only tools but keeps read-only-annotated tools."""
     assert not mcp_config
     monkeypatch.setenv(MCP_READONLY_MODE_ENV_VAR, "1")
-    for tool_name in ("run_cloud_sync", "cancel_cloud_sync"):
+    for tool_name in ("run_connection_sync", "cancel_connection_sync"):
         tool = _policy_tool(tool_name, pipeline_change=False)
         assert not _tool_utils.airbyte_readonly_mode_filter(tool, APP)
     read_only_tool = _policy_tool(
-        "list_cloud_workspaces", pipeline_change=False, read_only_hint=True
+        "list_workspaces", pipeline_change=False, read_only_hint=True
     )
     assert _tool_utils.airbyte_readonly_mode_filter(read_only_tool, APP)
 
@@ -408,7 +408,7 @@ def test_local_tools_unaffected_by_readonly_and_destructive_policies(
     # No `ToolPolicy` registration: the tools exercise the real default
     # (non-read-only -> pipeline_change=True).
     destructive_local_tool = _policy_tool(
-        "destination_smoke_test",
+        "run_local_destination_smoke_test",
         pipeline_change=None,
         destructive_hint=True,
         module="local",
@@ -442,5 +442,5 @@ def visible(name: str) -> bool:
             tool, airbyte_app
         ) and _tool_utils.airbyte_destructive_tools_filter(tool, airbyte_app)
 
-    assert visible("destination_smoke_test")
-    assert not visible("run_cloud_sync")
+    assert visible("run_local_destination_smoke_test")
+    assert not visible("run_connection_sync")
diff --git a/tests/unit_tests/test_mcp_tool_utils.py b/tests/unit_tests/test_mcp_tool_utils.py
index 25532465d..cf3200da7 100644
--- a/tests/unit_tests/test_mcp_tool_utils.py
+++ b/tests/unit_tests/test_mcp_tool_utils.py
@@ -283,13 +283,13 @@ def test_server_tools_register_expected_policies() -> None:
         "execute_external_api_query",
         "execute_external_sql_query",
         "execute_external_search_query",
-        "get_cloud_search_status",
+        "get_search_index_status",
         "get_external_access_skill_docs",
     )
     for tool_name in external_tools:
         assert _TOOL_POLICIES[tool_name].external_access
 
-    for tool_name in ("run_cloud_sync", "cancel_cloud_sync"):
+    for tool_name in ("run_connection_sync", "cancel_connection_sync"):
         assert _TOOL_POLICIES[tool_name].pipeline_change is False
 
     insiders_tools = {
@@ -417,11 +417,11 @@ def dedent_docstring_tool(value: str) -> None:
 @pytest.mark.parametrize(
     "tool_name",
     [
-        "validate_connector_config",
-        "list_dotenv_secrets",
-        "list_source_streams",
-        "get_stream_previews",
-        "run_sql_query",
+        "validate_local_connector_config",
+        "list_local_dotenv_secrets",
+        "list_local_source_streams",
+        "get_local_stream_previews",
+        "run_local_sql_query",
     ],
 )
 def test_local_tool_descriptions_are_dedented(tool_name: str) -> None:
diff --git a/tests/unit_tests/test_mcp_trusted_execution.py b/tests/unit_tests/test_mcp_trusted_execution.py
index acfa311d7..21cf57690 100644
--- a/tests/unit_tests/test_mcp_trusted_execution.py
+++ b/tests/unit_tests/test_mcp_trusted_execution.py
@@ -335,16 +335,24 @@ async def call_publish() -> object:
         lambda: local._get_mcp_source("source-faker", manifest_path=None),
         id="_get_mcp_source",
     ),
-    pytest.param(lambda: local.list_cached_streams(), id="list_cached_streams"),
-    pytest.param(lambda: local.describe_default_cache(), id="describe_default_cache"),
-    pytest.param(lambda: local.run_sql_query("SELECT 1", 10), id="run_sql_query"),
-    pytest.param(lambda: local.list_dotenv_secrets(), id="list_dotenv_secrets"),
     pytest.param(
-        lambda: local.list_connector_config_secrets("source-faker"),
-        id="list_connector_config_secrets",
+        lambda: local.list_local_cached_streams(), id="list_local_cached_streams"
     ),
     pytest.param(
-        lambda: local.destination_smoke_test(
+        lambda: local.describe_local_default_cache(), id="describe_local_default_cache"
+    ),
+    pytest.param(
+        lambda: local.run_local_sql_query("SELECT 1", 10), id="run_local_sql_query"
+    ),
+    pytest.param(
+        lambda: local.list_local_dotenv_secrets(), id="list_local_dotenv_secrets"
+    ),
+    pytest.param(
+        lambda: local.list_local_connector_config_secrets("source-faker"),
+        id="list_local_connector_config_secrets",
+    ),
+    pytest.param(
+        lambda: local.run_local_destination_smoke_test(
             "destination-dev-null",
             None,
             None,
@@ -356,7 +364,7 @@ async def call_publish() -> object:
             None,
             False,
         ),
-        id="destination_smoke_test",
+        id="run_local_destination_smoke_test",
     ),
 ]
 
@@ -391,7 +399,7 @@ def run_sql_query(self, sql_query: str, max_records: int) -> None:
     local.logger.addHandler(caplog.handler)
     try:
         caplog.set_level(logging.WARNING, logger=local.logger.name)
-        local.run_sql_query("SELECT 1", 10)
+        local.run_local_sql_query("SELECT 1", 10)
     finally:
         local.logger.removeHandler(caplog.handler)