Proposal: Consolidate MCP Tools Using Resource-Verb Pattern (43 → ~11 tools)

## Proposal: Consolidate MCP Tools Using Resource-Verb Pattern

### Summary

Refactor the PyAirbyte MCP server's 43 granular tools into ~11 consolidated tools using a `resource(verb, ...)` pattern. This reduces context window consumption and improves LLM tool selection accuracy.

### Motivation

The current MCP server exposes **43 separate tools** across 3 modules. Each tool definition consumes context tokens when loaded by an AI agent. Research from multiple sources confirms that tool overload degrades LLM performance:

- [Anthropic Engineering](https://www.anthropic.com/engineering/code-execution-with-mcp) (Nov 2025): Code execution reduced token usage from 150,000 to 2,000 tokens (98.7% reduction)
- [EclipseSource](https://eclipsesource.com/blogs/2026/01/22/mcp-context-overload/) (Jan 2026): Tool definitions consume >20% of context window before the agent starts working
- [Synaptic Labs - Meta-Tool Pattern](https://blog.synapticlabs.ai/bounded-context-packs-meta-tool-pattern) (Jan 2026): Reducing from 33 tools to 2 meta-tools cut startup cost from ~8,000 tokens to ~600 tokens
- [MCPBundles - Six-Tool Pattern](https://www.mcpbundles.com/blog/mcp-tool-design-pattern) (Oct 2025): Consolidating 18 tools into 6 tools with rich parameters
- [Klavis AI - "Less is More"](https://www.klavis.ai/blog/less-is-more-mcp-design-patterns-for-ai-agents) (Oct 2025): Workflow-based design over API-mirroring
- [Anthropic - Writing Effective Tools](https://www.anthropic.com/engineering/writing-tools-for-agents) (Sep 2025): Namespacing tools, consolidating by resource domain

### Current Tool Inventory (43 tools)

Show/Hide Full Tool List

**`registry.py` (4 tools):**
- `list_connectors`
- `get_connector_info`
- `get_api_docs_urls`
- `get_connector_version_history`

**`local.py` (11 tools):**
- `validate_connector_config`
- `list_connector_config_secrets`
- `list_dotenv_secrets`
- `list_source_streams`
- `get_source_stream_json_schema`
- `read_source_stream_records`
- `get_stream_previews`
- `sync_source_to_cache`
- `list_cached_streams`
- `describe_default_cache`
- `run_sql_query`

**`cloud.py` (28 tools):**
- `deploy_source_to_cloud`
- `deploy_destination_to_cloud`
- `deploy_noop_destination_to_cloud`
- `create_connection_on_cloud`
- `run_cloud_sync`
- `check_airbyte_cloud_workspace`
- `get_cloud_sync_status`
- `list_cloud_sync_jobs`
- `list_deployed_cloud_source_connectors`
- `list_deployed_cloud_destination_connectors`
- `describe_cloud_source`
- `describe_cloud_destination`
- `describe_cloud_connection`
- `get_cloud_sync_logs`
- `list_deployed_cloud_connections`
- `list_cloud_workspaces`
- `describe_cloud_organization`
- `publish_custom_source_definition`
- `list_custom_source_definitions`
- `get_custom_source_definition`
- `update_custom_source_definition`
- `permanently_delete_custom_source_definition`
- `permanently_delete_cloud_source`
- `permanently_delete_cloud_destination`
- `permanently_delete_cloud_connection`
- `rename_cloud_source`
- `update_cloud_source_config`
- `rename_cloud_destination`
- `update_cloud_destination_config`
- `rename_cloud_connection`
- `set_cloud_connection_table_prefix`
- `set_cloud_connection_selected_streams`
- `update_cloud_connection`
- `get_connection_artifact`

### Proposed Consolidated Mapping (43 → ~11 tools)

| Consolidated Tool | Verbs | Current Tools Absorbed | Count |
|---|---|---|---|
| `connector_registry(verb, ...)` | `list`, `get_info`, `get_api_docs`, `get_version_history` | `list_connectors`, `get_connector_info`, `get_api_docs_urls`, `get_connector_version_history` | 4→1 |
| `cloud_source(verb, ...)` | `deploy`, `list`, `describe`, `rename`, `update_config`, `delete` | `deploy_source_to_cloud`, `list_deployed_cloud_source_connectors`, `describe_cloud_source`, `rename_cloud_source`, `update_cloud_source_config`, `permanently_delete_cloud_source` | 6→1 |
| `cloud_destination(verb, ...)` | `deploy`, `list`, `describe`, `rename`, `update_config`, `delete` | `deploy_destination_to_cloud`, `deploy_noop_destination_to_cloud`, `list_deployed_cloud_destination_connectors`, `describe_cloud_destination`, `rename_cloud_destination`, `update_cloud_destination_config`, `permanently_delete_cloud_destination` | 7→1 |
| `cloud_connection(verb, ...)` | `create`, `list`, `describe`, `rename`, `update`, `delete`, `set_table_prefix`, `set_selected_streams`, `get_artifact` | `create_connection_on_cloud`, `list_deployed_cloud_connections`, `describe_cloud_connection`, `rename_cloud_connection`, `update_cloud_connection`, `permanently_delete_cloud_connection`, `set_cloud_connection_table_prefix`, `set_cloud_connection_selected_streams`, `get_connection_artifact` | 9→1 |
| `cloud_sync(verb, ...)` | `run`, `get_status`, `list_jobs`, `get_logs` | `run_cloud_sync`, `get_cloud_sync_status`, `list_cloud_sync_jobs`, `get_cloud_sync_logs` | 4→1 |
| `cloud_workspace(verb, ...)` | `check`, `list` | `check_airbyte_cloud_workspace`, `list_cloud_workspaces` | 2→1 |
| `cloud_organization(verb, ...)` | `describe` | `describe_cloud_organization` | 1→1 |
| `custom_source_definition(verb, ...)` | `publish`, `list`, `get`, `update`, `delete` | `publish_custom_source_definition`, `list_custom_source_definitions`, `get_custom_source_definition`, `update_custom_source_definition`, `permanently_delete_custom_source_definition` | 5→1 |
| `local_source(verb, ...)` | `validate_config`, `list_streams`, `get_schema`, `read_records`, `get_previews`, `sync_to_cache` | `validate_connector_config`, `list_source_streams`, `get_source_stream_json_schema`, `read_source_stream_records`, `get_stream_previews`, `sync_source_to_cache` | 6→1 |
| `local_cache(verb, ...)` | `list_streams`, `describe`, `run_sql` | `list_cached_streams`, `describe_default_cache`, `run_sql_query` | 3→1 |
| `secrets(verb, ...)` | `list_connector_secrets`, `list_dotenv` | `list_connector_config_secrets`, `list_dotenv_secrets` | 2→1 |

### Before/After Examples

Show/Hide Example: Cloud Source (6→1)

**Before (6 separate tools):**
```python
deploy_source_to_cloud(source_name, source_connector_name, config, ...)
list_deployed_cloud_source_connectors(name_contains, max_items_limit, ...)
describe_cloud_source(source_id, ...)
rename_cloud_source(source_id, name, ...)
update_cloud_source_config(source_id, config, ...)
permanently_delete_cloud_source(source_id, name, ...)
```

**After (1 consolidated tool):**
```python
cloud_source(
verb: "deploy" | "list" | "describe" | "rename" | "update_config" | "delete",
source_id: str | None, # required for describe/rename/update_config/delete
source_name: str | None, # required for deploy, rename
source_connector_name: str | None, # required for deploy
config: dict | str | None, # for deploy, update_config
name_contains: str | None, # for list
max_items_limit: int | None, # for list
workspace_id: str | None,
...
)
```

Show/Hide Example: Cloud Connection (9→1)

**Before (9 separate tools):**
```python
create_connection_on_cloud(connection_name, source_id, destination_id, selected_streams, ...)
list_deployed_cloud_connections(name_contains, with_connection_status, ...)
describe_cloud_connection(connection_id, ...)
rename_cloud_connection(connection_id, name, ...)
update_cloud_connection(connection_id, enabled, cron_expression, ...)
permanently_delete_cloud_connection(connection_id, name, ...)
set_cloud_connection_table_prefix(connection_id, prefix, ...)
set_cloud_connection_selected_streams(connection_id, stream_names, ...)
get_connection_artifact(connection_id, artifact_type, ...)
```

**After (1 consolidated tool):**
```python
cloud_connection(
verb: "create" | "list" | "describe" | "rename" | "update" | "delete"
| "set_table_prefix" | "set_selected_streams" | "get_artifact",
connection_id: str | None,
connection_name: str | None,
source_id: str | None,
destination_id: str | None,
selected_streams: list[str] | None,
artifact_type: "state" | "catalog" | None,
...
)
```

Show/Hide Example: Connector Registry (4→1)

**Before (4 separate tools):**
```python
list_connectors(keyword_filter, connector_type_filter, install_types)
get_connector_info(connector_name)
get_api_docs_urls(connector_name)
get_connector_version_history(connector_name, num_versions_to_validate)
```

**After (1 consolidated tool):**
```python
connector_registry(
verb: "list" | "get_info" | "get_api_docs" | "get_version_history",
connector_name: str | None, # required for get_info/get_api_docs/get_version_history
keyword_filter: str | None, # for list
connector_type_filter: str | None, # for list
install_types: list[str] | None, # for list
num_versions_to_validate: int, # for get_version_history
)
```

### Gotchas to Watch For

1. **Don't lose discoverability** — Put the verb enum and descriptions in the tool's schema so the model sees all capabilities upfront.
2. **Enum explosion** — If a verb parameter grows to 15+ values with completely different parameter schemas per verb, consider splitting at natural domain boundaries instead.
3. **Parameter schema divergence** — When different verbs need very different params, either use `oneOf`/conditional schemas (which some clients handle poorly) or make everything optional (which loses validation). Test with target clients.
4. **Client compatibility** — Verify consolidated tool schemas work across Claude, ChatGPT, Cursor, etc.
5. **Error attribution** — With consolidated tools, make sure the verb is always logged prominently so errors are easy to trace.

### Open Questions

- Should `cloud_organization` (1 verb) be merged into `cloud_workspace`?
- Should `deploy_noop_destination_to_cloud` be a special case under `cloud_destination(verb="deploy")` or remain standalone?
- Should `secrets` (2 verbs) remain as 2 separate small tools or consolidate?

### References

- Devin research session: https://app.devin.ai/sessions/82f92fc2616c460ebafd06d75a885e57
- Original discussion thread in Slack: `#ask-devin-ai`
- Requested by: @aaronsteers