setup_connector
Set up or delete exchange connectors for automated trading with progressive disclosure: list exchanges, show required credentials, select account, and confirm override.
Instructions
Setup or delete an exchange connector for an account with credentials using progressive disclosure.
This tool guides you through the entire process of connecting an exchange with a four-step flow:
1. No parameters → List available exchanges
2. Connector only → Show required credential fields
3. Connector + credentials, no account → Select account from available accounts
4. All parameters → Connect the exchange (with override confirmation if needed)
Delete flow (action="delete"):
1. action="delete" only → List all accounts and their configured connectors
2. action="delete" + connector → Show which accounts have this connector configured
3. action="delete" + connector + account → Delete the credential
Args:
action: Action to perform. 'setup' (default) to add/update credentials, 'delete' to remove credentials.
connector: Exchange connector name (e.g., 'binance', 'binance_perpetual'). Leave empty to list available connectors.
credentials: Credentials object with required fields for the connector. Leave empty to see required fields first.
account: Account name to add credentials to. If not provided, prompts for account selection.
confirm_override: Explicit confirmation to override existing connector. Required when connector already exists.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| action | No | ||
| connector | No | ||
| credentials | No | ||
| account | No | ||
| confirm_override | No |
Implementation Reference
- hummingbot_mcp/server.py:64-100 (registration)Registration of the setup_connector tool as an MCP tool via @mcp.tool() decorator. It parses parameters into a SetupConnectorRequest, calls the implementation, and formats the result.
@mcp.tool() @handle_errors("setup/delete connector") async def setup_connector( action: Literal["setup", "delete"] | None = None, connector: str | None = None, credentials: dict[str, Any] | None = None, account: str | None = None, confirm_override: bool | None = None, ) -> str: """Setup or delete an exchange connector for an account with credentials using progressive disclosure. This tool guides you through the entire process of connecting an exchange with a four-step flow: 1. No parameters → List available exchanges 2. Connector only → Show required credential fields 3. Connector + credentials, no account → Select account from available accounts 4. All parameters → Connect the exchange (with override confirmation if needed) Delete flow (action="delete"): 1. action="delete" only → List all accounts and their configured connectors 2. action="delete" + connector → Show which accounts have this connector configured 3. action="delete" + connector + account → Delete the credential Args: action: Action to perform. 'setup' (default) to add/update credentials, 'delete' to remove credentials. connector: Exchange connector name (e.g., 'binance', 'binance_perpetual'). Leave empty to list available connectors. credentials: Credentials object with required fields for the connector. Leave empty to see required fields first. account: Account name to add credentials to. If not provided, prompts for account selection. confirm_override: Explicit confirmation to override existing connector. Required when connector already exists. """ request = SetupConnectorRequest( action=action, connector=connector, credentials=credentials, account=account, confirm_override=confirm_override, ) client = await hummingbot_client.get_client() result = await setup_connector_impl(client, request) return format_connector_result(result) - hummingbot_mcp/tools/account.py:25-250 (handler)Core implementation of setup_connector logic. Handles both setup flow (list exchanges, show config, select account, connect) and delete flow (list, select account, delete) with progressive disclosure.
async def setup_connector(client: Any, request: SetupConnectorRequest) -> dict[str, Any]: """Setup or delete an exchange connector with credentials using progressive disclosure. Setup flow: 1. No connector -> List available exchanges 2. Connector only -> Show required credential fields 3. Connector + credentials, no account -> Select account from available accounts 4. All parameters -> Connect the exchange (with override confirmation if needed) Delete flow: 1. action="delete" only -> List accounts and their configured connectors 2. action="delete" + connector -> Show which accounts have this connector 3. action="delete" + connector + account -> Delete the credential """ flow_stage = request.get_flow_stage() # ============================ # Delete Flow # ============================ if flow_stage == "delete_list": # List all accounts and their configured connectors accounts = await client.accounts.list_accounts() credentials_tasks = [ client.accounts.list_account_credentials(account_name=account_name) for account_name in accounts ] credentials = await asyncio.gather(*credentials_tasks) account_connectors = {} for account, creds in zip(accounts, credentials): account_connectors[account] = creds if creds else [] return { "action": "delete_list", "message": "Configured connectors by account:", "account_connectors": account_connectors, "next_step": "Call again with action='delete' and 'connector' to select which connector to remove", "example": "Use action='delete', connector='binance' to remove Binance credentials", } elif flow_stage == "delete_select_account": # Show which accounts have this connector configured accounts = await client.accounts.list_accounts() credentials_tasks = [ client.accounts.list_account_credentials(account_name=account_name) for account_name in accounts ] credentials = await asyncio.gather(*credentials_tasks) matching_accounts = [] for account, creds in zip(accounts, credentials): if request.connector in (creds or []): matching_accounts.append(account) if not matching_accounts: return { "action": "delete_not_found", "message": f"Connector '{request.connector}' is not configured on any account", "connector": request.connector, "next_step": "Use action='delete' without a connector to see all configured connectors", } return { "action": "delete_select_account", "message": f"Connector '{request.connector}' is configured on the following accounts:", "connector": request.connector, "accounts": matching_accounts, "default_account": settings.default_account, "next_step": "Call again with 'account' to specify which account to delete from", "example": f"Use action='delete', connector='{request.connector}', " f"account='{matching_accounts[0]}' to delete", } elif flow_stage == "delete": # Actually delete the credential account_name = request.get_account_name() # Verify the connector exists before deleting connector_exists = await _check_existing_connector(client, account_name, request.connector) if not connector_exists: return { "action": "delete_not_found", "message": f"Connector '{request.connector}' is not configured on account '{account_name}'", "account": account_name, "connector": request.connector, "next_step": "Use action='delete' without parameters to see all configured connectors", } try: await client.accounts.delete_credential( account_name=account_name, connector_name=request.connector, ) return { "action": "credentials_deleted", "message": f"Successfully deleted {request.connector} credentials from account {account_name}", "account": account_name, "connector": request.connector, "next_step": "Use setup_connector() to see remaining configured connectors", } except Exception as e: raise ToolError(f"Failed to delete credentials for {request.connector}: {str(e)}") # ============================ # Setup Flow # ============================ elif flow_stage == "select_account": # Step 2.5: List available accounts for selection (after connector and credentials are provided) accounts = await client.accounts.list_accounts() return { "action": "select_account", "message": f"Ready to connect {request.connector}. Please select an account:", "connector": request.connector, "accounts": accounts, "default_account": settings.default_account, "next_step": "Call again with 'account' parameter to specify which account to use", "example": f"Use account='{settings.default_account}' to use the default account, or choose from " f"the available accounts above", } elif flow_stage == "list_exchanges": # Step 1: List available connectors connectors = await client.connectors.list_connectors() # Handle both string and object responses from the API connector_names = [] for c in connectors: if isinstance(c, str): connector_names.append(c) elif hasattr(c, "name"): connector_names.append(c.name) else: connector_names.append(str(c)) current_accounts_str = "Current accounts: " accounts = await client.accounts.list_accounts() credentials_tasks = [client.accounts.list_account_credentials(account_name=account_name) for account_name in accounts] credentials = await asyncio.gather(*credentials_tasks) for account, creds in zip(accounts, credentials): current_accounts_str += f"{account}: {creds}), " return { "action": "list_connectors", "message": "Available exchange connectors:", "connectors": connector_names, "total_connectors": len(connector_names), "current_accounts": current_accounts_str.strip(", "), "next_step": "Call again with 'connector' parameter to see required credentials for a specific exchange", "example": "Use connector='binance' to see Binance setup requirements", } elif flow_stage == "show_config": # Step 2: Show required credential fields for the connector try: config_fields = await client.connectors.get_config_map(request.connector) # Build a dictionary from the list of field names credentials_dict = {field: f"your_{field}" for field in config_fields} return { "action": "show_config_map", "connector": request.connector, "required_fields": config_fields, "next_step": "Call again with 'credentials' parameter containing the required fields", "example": f"Use credentials={credentials_dict} to connect", } except Exception as e: raise ToolError(f"Failed to get configuration for connector '{request.connector}': {str(e)}") elif flow_stage == "connect": # Step 3: Actually connect the exchange with provided credentials account_name = request.get_account_name() # Check if connector already exists connector_exists = await _check_existing_connector(client, account_name, request.connector) if connector_exists and request.requires_override_confirmation(): return { "action": "requires_confirmation", "message": f"WARNING: Connector '{request.connector}' already exists for account '{account_name}'", "account": account_name, "connector": request.connector, "warning": "Adding credentials will override the existing connector configuration", "next_step": "To proceed with overriding, add 'confirm_override': true to your request", "example": "Use confirm_override=true along with your credentials to override the existing connector", } if connector_exists and not request.confirm_override: return { "action": "override_rejected", "message": f"Cannot override existing connector {request.connector} without explicit confirmation", "account": account_name, "connector": request.connector, "next_step": "Set confirm_override=true to override the existing connector", } # Remove force_override from credentials before sending to API credentials_to_send = dict(request.credentials) if "force_override" in credentials_to_send: del credentials_to_send["force_override"] try: await client.accounts.add_credential( account_name=account_name, connector_name=request.connector, credentials=credentials_to_send ) action_type = "credentials_overridden" if connector_exists else "credentials_added" message_action = "overridden" if connector_exists else "connected" return { "action": action_type, "message": f"Successfully {message_action} {request.connector} exchange to account {account_name}", "account": account_name, "connector": request.connector, "credentials_count": len(credentials_to_send), "was_existing": connector_exists, "next_step": "Exchange is now ready for trading. Use get_account_state to verify the connection.", } except Exception as e: raise ToolError(f"Failed to add credentials for {request.connector}: {str(e)}") else: raise ToolError(f"Unknown flow stage: {flow_stage}") - hummingbot_mcp/schemas.py:20-136 (schema)Pydantic request model SetupConnectorRequest with fields (action, account, connector, credentials, confirm_override) and flow logic methods: get_account_name(), get_flow_stage(), requires_override_confirmation(), plus field validators.
class SetupConnectorRequest(BaseModel): """Request model for setting up exchange connectors with progressive disclosure. This model supports setup and delete flows: Setup flow (action=None or action="setup"): 1. No parameters -> List available exchanges 2. Connector only -> Show required credential fields 3. Connector + credentials, no account -> Select account from available accounts 4. All parameters -> Connect the exchange (with override confirmation if needed) Delete flow (action="delete"): 1. action="delete" only -> List accounts and their configured connectors 2. action="delete" + connector -> Show which accounts have this connector 3. action="delete" + connector + account -> Delete the credential """ action: Literal["setup", "delete"] | None = Field( default=None, description="Action to perform. 'setup' (default) to add/update credentials, 'delete' to remove credentials.", ) account: str | None = Field( default=None, description="Account name to add credentials to. If not provided, uses the default account." ) connector: str | None = Field( default=None, description="Exchange connector name (e.g., 'binance', 'coinbase_pro'). Leave empty to list available connectors.", examples=["binance", "coinbase_pro", "kraken", "gate_io"], ) credentials: dict[str, Any] | None = Field( default=None, description="Credentials object with required fields for the connector. Leave empty to see required fields first.", examples=[ {"binance_api_key": "your_api_key", "binance_secret_key": "your_secret"}, { "coinbase_pro_api_key": "your_key", "coinbase_pro_secret_key": "your_secret", "coinbase_pro_passphrase": "your_passphrase", }, ], ) confirm_override: bool | None = Field( default=None, description="Explicit confirmation to override existing connector. Required when connector already exists.", ) @field_validator("connector") @classmethod def validate_connector_name(cls, v: str | None) -> str | None: """Validate connector name format if provided""" if v is not None: # Convert to lowercase and replace spaces/hyphens with underscores v = v.lower().replace(" ", "_").replace("-", "_") # Basic validation - should be alphanumeric with underscores if not v.replace("_", "").isalnum(): raise ValueError("Connector name should contain only letters, numbers, and underscores") return v @field_validator("credentials") @classmethod def validate_credentials(cls, v: dict[str, Any] | None) -> dict[str, Any] | None: """Validate credentials format if provided""" if v is not None: if not isinstance(v, dict): raise ValueError("Credentials must be a dictionary/object") if not v: # Empty dict raise ValueError("Credentials cannot be empty. Omit the field to see required fields.") # Check that all values are strings (typical for API credentials) # except for force_override which can be boolean for key, value in v.items(): if key == "force_override": if not isinstance(value, bool): raise ValueError("'force_override' must be a boolean (true/false)") else: if not isinstance(value, str): raise ValueError(f"Credential '{key}' must be a string") if not value.strip(): # Empty or whitespace-only raise ValueError(f"Credential '{key}' cannot be empty") return v def get_account_name(self) -> str: """Get account name with fallback to default""" return self.account or settings.default_account def get_flow_stage(self) -> str: """Determine which stage of the setup/delete flow we're in""" if self.action == "delete": if self.connector is None: return "delete_list" elif self.account is None: return "delete_select_account" else: return "delete" if self.connector is None: return "list_exchanges" elif self.credentials is None: return "show_config" elif self.account is None: return "select_account" else: return "connect" def requires_override_confirmation(self) -> bool: """Check if this request needs override confirmation""" return self.credentials is not None and self.confirm_override is None - Formatting helper format_connector_result that converts the dict result from setup_connector into a human-readable string for all flow stages.
def format_connector_result(result: dict[str, Any]) -> str: """Format the result from setup_connector/delete_connector into a human-readable string.""" result_action = result.get("action", "") if result_action == "list_connectors": connectors = result.get("connectors", []) connector_lines = [] for i in range(0, len(connectors), 4): line = " ".join(f"{c:25}" for c in connectors[i:i+4]) connector_lines.append(line) return ( f"Available Exchange Connectors ({result.get('total_connectors', 0)} total):\n\n" + "\n".join(connector_lines) + "\n\n" f"{result.get('current_accounts', '')}\n\n" f"Next Step: {result.get('next_step', '')}\n" f"Example: {result.get('example', '')}" ) elif result_action == "show_config_map": fields = result.get("required_fields", []) return ( f"Required Credentials for {result.get('connector', '')}:\n\n" f"Fields needed:\n" + "\n".join(f" - {field}" for field in fields) + "\n\n" f"Next Step: {result.get('next_step', '')}\n" f"Example: {result.get('example', '')}" ) elif result_action == "select_account": accounts = result.get("accounts", []) return ( f"{result.get('message', '')}\n\n" f"Available Accounts:\n" + "\n".join(f" - {acc}" for acc in accounts) + "\n\n" f"Default Account: {result.get('default_account', '')}\n\n" f"Next Step: {result.get('next_step', '')}\n" f"Example: {result.get('example', '')}" ) elif result_action == "requires_confirmation": return ( f"\u26a0\ufe0f {result.get('message', '')}\n\n" f"Account: {result.get('account', '')}\n" f"Connector: {result.get('connector', '')}\n" f"Warning: {result.get('warning', '')}\n\n" f"Next Step: {result.get('next_step', '')}\n" f"Example: {result.get('example', '')}" ) elif result_action == "override_rejected": return ( f"\u274c {result.get('message', '')}\n\n" f"Account: {result.get('account', '')}\n" f"Connector: {result.get('connector', '')}\n\n" f"Next Step: {result.get('next_step', '')}" ) elif result_action in ["credentials_added", "credentials_overridden"]: return ( f"\u2705 {result.get('message', '')}\n\n" f"Account: {result.get('account', '')}\n" f"Connector: {result.get('connector', '')}\n" f"Credentials Count: {result.get('credentials_count', 0)}\n" f"Was Existing: {result.get('was_existing', False)}\n\n" f"Next Step: {result.get('next_step', '')}" ) # Delete flow responses elif result_action == "delete_list": account_connectors = result.get("account_connectors", {}) lines = [result.get("message", "")] for acc, conns in account_connectors.items(): conns_str = ", ".join(conns) if conns else "(none)" lines.append(f" - {acc}: {conns_str}") lines.append("") lines.append(f"Next Step: {result.get('next_step', '')}") lines.append(f"Example: {result.get('example', '')}") return "\n".join(lines) elif result_action == "delete_select_account": accounts = result.get("accounts", []) return ( f"{result.get('message', '')}\n\n" f"Accounts:\n" + "\n".join(f" - {acc}" for acc in accounts) + "\n\n" f"Default Account: {result.get('default_account', '')}\n\n" f"Next Step: {result.get('next_step', '')}\n" f"Example: {result.get('example', '')}" ) elif result_action == "delete_not_found": return ( f"\u274c {result.get('message', '')}\n\n" f"Next Step: {result.get('next_step', '')}" ) elif result_action == "credentials_deleted": return ( f"\u2705 {result.get('message', '')}\n\n" f"Account: {result.get('account', '')}\n" f"Connector: {result.get('connector', '')}\n\n" f"Next Step: {result.get('next_step', '')}" ) # Fallback for unknown actions return f"Connector Result: {result}" - Helper function _check_existing_connector used by setup_connector to check if a connector already exists for an account.
async def _check_existing_connector(client: Any, account_name: str, connector_name: str) -> bool: """Check if a connector already exists for the given account""" try: credentials = await client.accounts.list_account_credentials(account_name=account_name) return connector_name in credentials except Exception as e: logger.warning(f"Failed to check existing connector: {str(e)}")