Resource Connectors and Connector Groups API Guide
This guide provides Python client samples for the Cisco Secure Access Resource Connectors API.
Note: Your Secure Access API key must have the permissions to read and write on the deployments.resourceconnectors key scope. For more information about the API key scopes, see Secure Access OAuth 2.0 Scopes.
First get your Secure Access API key, set up your environment, and install the Secure Access API client. For more information, see Samples Overview.
Note: You must deploy resource connectors in your cloud environment. You can not deploy a resource connector through the Resource Connector API. For more information about deploying resource connectors, see Add Connectors to a Connector Group.
Run the Script
- Copy the script to a local file called
main.py. Locate the script in your environment in a directory above theciscodirectory. - Run
python3 main.py.
main.py
"""
Copyright (c) 2025 Cisco and/or its affiliates.
This software is licensed to you under the terms of the Cisco Sample
Code License, Version 1.1 (the "License"). You may obtain a copy of the
License at
https://developer.cisco.com/docs/licenses
All use of the material herein must be in accordance with the terms of
the License. All rights not expressly granted by the License are
reserved. Unless required by applicable law or agreed to separately in
writing, software distributed under the License is distributed on an "AS
IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
or implied.
"""
import requests
from requests_toolbelt import MultipartEncoder
import json
import os
from dotenv import load_dotenv
from cisco.secure_access import API
from cisco.secure_access import deployments
from cisco.secure_access import GET
from cisco.secure_access import PATCH
from cisco.secure_access import POST
from cisco.secure_access import PUT
from cisco.secure_access import DELETE
from cisco.secure_access import POST_MULTIPART_FORM_DATA
from cisco.secure_access import token_url
from cisco.secure_access import client_id
from cisco.secure_access import client_secret
# Resource Connectors and Groups API endpoints
connector_groups_endpoint = "connectorGroups"
connector_groups_details_endpoint = "connectorGroups/{}"
connectors_endpoint = "connectorAgents"
connectors_details_endpoint = "connectorAgents/{}"
connector_groups_counts_endpoint = "connectorGroups/counts"
connectors_counts_endpoint = "connectorAgents/counts"
load_dotenv()
def get_connector_groups(api):
''' Get Connector Groups. '''
try:
# Get Resource Connector Groups in the organization
response = api.Query(deployments, connector_groups_endpoint, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {connector_groups_endpoint}, {response.json()}")
return response.json()
else:
print(f"Failed to get the Resource Connector Groups. Status code: {response.status_code}, Response: {response.text}.")
return None
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}.")
def post_connector_group(api, name=None, location=None, environment=None):
''' Create a Resource Connector Group. '''
try:
if name is None or location is None or environment is None:
raise ValueError("name, location, and environment are required to create the Resource Connector Group.")
# Prepare the payload
payload = {
"name": name,
"location": location,
"environment": environment
}
# Create a Resource Connector Group
response = api.Query(deployments, connector_groups_endpoint, POST, payload)
# Check the response status
if response.status_code == 201:
print(f"Success. POST {connector_groups_endpoint}, {response.json()}")
return response.json()
else:
print(f"Failed to create the Resource Connector Group {name}. Status code: {response.status_code}, Response: {response.text}.")
return None
except Exception as e:
print(f"An error occurred: {e}.")
def get_connector_group(api, id):
''' Get the properties of the Connector Group. '''
try:
if id is None:
raise ValueError("id is required to get the Resource Connector Group.")
url = connector_groups_details_endpoint.format(id)
# Get the properties for the Resource Connector Group
response = api.Query(deployments, url, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {url}, {response.json()}.")
return response.json()
else:
print(f"Failed to get the Resource Connector Group {id}. Status code: {response.status_code}, Response: {response.text}.")
return
except Exception as e:
print(f"An error occurred: {e}.")
def put_connector_group(api, id, name=None, location=None, environment=None):
''' Update the properties of the Connector Group. '''
try:
if id is None:
raise ValueError("id is required to get the Resource Connector Group.")
if name is None or location is None or environment is None:
raise ValueError("name, location, or environment are required to update the Resource Connector Group.")
# Prepare the payload
payload = {
"name": name,
"location": location,
"environment": environment
}
url = connector_groups_details_endpoint.format(id)
# Update the properties for the Resource Connector Group
response = api.Query(deployments, url, PUT, payload)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. PUT {url}, {response.json()}.")
return response.json()
else:
print(f"Failed to update the Resource Connector {id}. Status code: {response.status_code}, Response: {response.text}.")
return
except Exception as e:
print(f"An error occurred: {e}.")
def patch_connector_group(api, id, data=None):
''' Set the properties of the Connector Group. '''
try:
if id is None:
raise ValueError("id is required to get the Resource Connector Group.")
if data is None:
raise ValueError("data is required to set the properties of the Resource Connector Group.")
url = connector_groups_details_endpoint.format(id)
# Set the properties for the Resource Connector Group
response = api.Query(deployments, url, PATCH, data)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. PATCH {url}, {response.json()}.")
return response.json()
else:
print(f"Failed to update the Resource Connector Group {id}. Status code: {response.status_code}, Response: {response.text}.")
return
except Exception as e:
print(f"An error occurred: {e}.")
def get_connector_groups_counts(api):
''' Get Connector Groups counts. '''
try:
# Get the counts for the Resource Connector Groups.
response = api.Query(deployments, connector_groups_counts_endpoint, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {connector_groups_counts_endpoint}, {response.json()}")
return response.json()
else:
print(f"Failed to get counts for Resource Connector Groups. Status code: {response.status_code}, Response: {response.text}.")
return None
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}.")
def delete_connector_group(api, id):
''' Delete the Connector Group. '''
try:
if id is None:
raise ValueError("id is required to delete the Resource Connector Group.")
url = connector_groups_details_endpoint.format(id)
# Delete the Resource Connector Group
response = api.Query(deployments, url, DELETE)
# Check if the API request was successful
if response.status_code == 204:
print(f"No Content. DELETE {url}.")
else:
print(f"Failed to delete the Resource Connector Group {id}. Status code: {response.status_code}, Response: {response.text}.")
except Exception as e:
print(f"An error occurred: {e}.")
# Connectors
def get_connectors(api):
''' Get Connectors. '''
try:
# Get Resource Connectors in the organization
response = api.Query(deployments, connectors_endpoint, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {connectors_endpoint}, {response.json()}")
return response.json()
else:
print(f"Failed to get the Resource Connectors. Status code: {response.status_code}, Response: {response.text}.")
return None
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}.")
def get_connector(api, id):
''' Get the properties of the Connector. '''
try:
if id is None:
raise ValueError("id is required to get the Resource Connector.")
url = connectors_details_endpoint.format(id)
# Get the properties for the Resource Connector
response = api.Query(deployments, url, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {url}, {response.json()}.")
return response.json()
else:
print(f"Failed to get the Resource Connector {id}. Status code: {response.status_code}, Response: {response.text}.")
return
except Exception as e:
print(f"An error occurred: {e}.")
def patch_connector(api, id, data):
''' Set the properties of the Resource Connector. '''
try:
if id is None:
raise ValueError("id is required to get the Resource Connector.")
if data is None:
raise ValueError("data is required to set the properties of the Resource Connector.")
url = connectors_details_endpoint.format(id)
# Set the properties for the Resource Connector Group
response = api.Query(deployments, url, PATCH, data)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. PATCH {url}, {response.json()}.")
return response.json()
else:
print(f"Failed to update the Resource Connector {id}. Status code: {response.status_code}, Response: {response.text}.")
return
except Exception as e:
print(f"An error occurred: {e}.")
def get_connectors_counts(api):
''' Get Connectors counts. '''
try:
# Get the counts for the Resource Connectors.
response = api.Query(deployments, connectors_counts_endpoint, GET)
# Check if the API request was successful
if response.status_code == 200:
print(f"Success. GET {connectors_counts_endpoint}, {response.json()}")
return response.json()
else:
print(f"Failed to get counts for Resource Connectors. Status code: {response.status_code}, Response: {response.text}.")
return None
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}.")
def delete_connector(api, id):
''' Delete the Connector. '''
try:
if id is None:
raise ValueError("id is required to delete the Resource Connector.")
url = connectors_details_endpoint.format(id)
# Delete the Resource Connector
response = api.Query(deployments, url, DELETE)
# Check if the API request was successful
if response.status_code == 204:
print(f"No Content. DELETE {url}.")
else:
print(f"Failed to delete the Resource Connector {id}. Status code: {response.status_code}, Response: {response.text}.")
except Exception as e:
print(f"An error occurred: {e}.")
def main():
# Exit out if the required client_id or client_secret is not set
for var in ['API_KEY', 'API_SECRET', 'OUTPUT_DIR']:
if os.environ.get(var) == None:
print("Required environment variable: {} not set".format(var))
exit()
# Get an API token
api = API(token_url, client_id, client_secret)
try:
# get the resource connector groups in the organization
json_data = get_connector_groups(api)
# create a resource connector group
name = "Yosemite Connector Group1"
location = "us-west-2"
environment = "aws"
json_data = post_connector_group(api, name, location, environment)
# get a resource connector group
if 'id' in json_data:
connectorGroupId = json_data['id']
json_data = get_connector_group(api, connectorGroupId)
# update a resource connector group
name = 'NYC Connector Group1'
location = "us-east-2"
environment = "aws"
json_data = put_connector_group(api, connectorGroupId, name, location, environment)
# patch a resource connector group
data = [
{
"op": 'replace',
"path": "/provisioningKey",
"value": ""
}
]
json_data = patch_connector_group(api, connectorGroupId, data)
# get the counts of the resource connector groups' states
json_data = get_connector_groups_counts(api)
# get connectors
json_data = get_connectors(api)
# get a connector listed in the response
connectorId = 0
if len(json_data['data']) > 0:
connector = json_data['data'][0]
connectorId = connector['id']
# get a resource connector
if connectorId != 0:
json_data = get_connector(api, connectorId)
# patch a resource connector
data = [
{
"op": "replace",
"path": "/confirmed",
"value": True
}
]
json_data = patch_connector(api, connectorId, data)
# get the counts of the connectors' states
json_data = get_connectors_counts(api)
if connectorId != 0:
# delete a resource connector
json_data = delete_connector(api, connectorId)
# delete a resource connector group
json_data = delete_connector_group(api, connectorGroupId)
except Exception as e:
print(e)
if __name__ == "__main__":
main()