def get_action_schema(self, action: str) -> Dict[str, Any]:
    """Retrieves the input and output JSON schema for a given action in a connection.

    Args:
        action (str): The action name.

    Returns:
        tuple: A tuple containing (input_schema, output_schema).

    Raises:
        PermissionError: If there are credential issues.
        ValueError: If there's a request or processing error.
        Exception: For any other unexpected errors.
    """
    url = f"{self.connector_url}/v1/projects/{self.project}/locations/{self.location}/connections/{self.connection}/connectionSchemaMetadata:getAction?actionId={action}"

    response = self._execute_api_call(url)

    operation_id = response.json().get("name")

    if not operation_id:
      raise ValueError(f"Failed to get action schema for action: {action}")

    operation_response = self._poll_operation(operation_id)

    input_schema = operation_response.get("response", {}).get(
        "inputJsonSchema", {}
    )
    output_schema = operation_response.get("response", {}).get(
        "outputJsonSchema", {}
    )
    description = operation_response.get("response", {}).get("description", "")
    display_name = operation_response.get("response", {}).get("displayName", "")
    return {
        "inputSchema": input_schema,
        "outputSchema": output_schema,
        "description": description,
        "displayName": display_name,
    }
Retrieves the input and output JSON schema for a given action in a connection.

    Args:
        action (str): The action name.

    Returns:
        tuple: A tuple containing (input_schema, output_schema).

    Raises:
        PermissionError: If there are credential issues.
        ValueError: If there's a request or processing error.
        Exception: For any other unexpected errors.
    
src/google/adk/tools/application_integration_tool/clients/connections_client.py