Skip to main content

​
Defining a python tools expected credentials

When building Python tools that consume connections, you must explicitly define each expected connection in the expected_credentials array within the @tool decorator. If your Python file includes multiple tools, each tool must declare its own expected_credentials. This ensures that the tool can fetch the necessary connections during execution. Include all connections used directly within the tool function, as well as those used in any helper functions the tool calls. If you attempt to import a tool with a connection type that doesn’t match one of the expected types for the specified app_id, the orchestrate tools import command will fail and prevent the tool from being imported.
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.API_KEY_AUTH}
    ]
)
def my_tool():
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    connection_type = connections.connection_type(MY_APP_ID2)
    conn = connections.api_key_auth(MY_APP_ID2)
    url = conn.url
    headers['x-api-key'] = conn.api_key

    response = requests.post(
        url,
        headers=headers,
        json={}
    )
    response.raise_for_status()
    return response.json()['result']

​
Examples of fetching credentials per connection type

Here’s an example that shows how to look up connection credentials for each credential type:
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.BASIC_AUTH}
    ]
)
def my_tool():
    creds = connections.basic_auth(MY_APP_ID)
    base_url = creds.url
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
        auth=HTTPBasicAuth(creds.username, creds.password)
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.BEARER_TOKEN}
    ]
)
def my_tool():
    creds = connections.bearer_token(MY_APP_ID)
    base_url = creds.url
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': creds.token,
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.API_KEY_AUTH}
    ]
)
def my_tool():
    creds = connections.api_key_auth(MY_APP_ID)
    base_url = creds.url
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'x-api-key': creds.api_key
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.OAUTH2_CLIENT_CREDS}
    ]
)
def my_tool():
    creds = connections.oauth2_client_creds(MY_APP_ID)
    base_url = creds.'url'
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f"Bearer {creds.access_token}"
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.OAUTH2_AUTH_CODE}
    ]
)
def my_tool():
    creds = connections.oauth2_auth_code(MY_APP_ID)
    base_url = creds.'url'
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f"Bearer {creds.access_token}"
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
Coming soon
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.OAUTH2_PASSWORD}
    ]
)
def my_tool():
    creds = connections.oauth2_password(MY_APP_ID)
    base_url = creds.'url'
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f"Bearer {creds.access_token}"
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.OAUTH_ON_BEHALF_OF_FLOW}
    ]
)
def my_tool():
    creds = connections.oauth2_on_behalf_of(MY_APP_ID)
    base_url = creds.'url'
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'Authorization': f"Bearer {creds.access_token}"
    }
    payload = {}

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.KEY_VALUE}
    ]
)
def my_tool():
    creds = connections.key_value(MY_APP_ID)
    base_url = creds.get('url', 'http://api.example.com')
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json',
        'x-api-key': creds.get('api_key')
        'x-super-secret': creds.get('super_secret')
    }
    payload = {
        'super_duper_secret': creds.get('super_duper_secret', 'secret')
    }

    response = requests.post(
        url,
        headers=headers,
        json=payload,
    )
    response.raise_for_status()
    return response.json()['result']

​
Combining multiple connection types for a single application id

You can define multiple authentication types for a single app_id, which is useful when using different methods in different environments—for example, basic authentication in draft mode and OAuth in production. To support this, provide an array of credential types in the expected_credentials field instead of a single type. At runtime, your Python tool must determine which connection type to use by calling: 
connections.connection_type('my-app')
Make sure the tool handles disambiguation logic appropriately. If the connection type provided during import doesn’t match any of the expected types for the specified app_id, the orchestrate tools import command will fail and block the import.
PYTHON
from ibm_watsonx_orchestrate.agent_builder.tools import tool
from ibm_watsonx_orchestrate.agent_builder.connections import ConnectionType
from ibm_watsonx_orchestrate.run import connections

MY_APP_ID='app-id'
MY_APP_ID2='app-id-2'

@tool(
    expected_credentials=[
        {"app_id": MY_APP_ID, "type": ConnectionType.KEY_VALUE},
        {"app_id": MY_APP_ID2, "type": [ConnectionType.API_KEY_AUTH, ConnectionType.OAUTH2_CLIENT_CREDS]}
    ]
)
def my_tool():
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
        # key value connections hold any arbitrary dictionary of keys and values
        'x-client-id': connections.key_value(MY_APP_ID)['client-id']
    }


    url = None
    connection_type = connections.connection_type(MY_APP_ID2)
    if connection_type == ConnectionType.API_KEY_AUTH:
        conn = connections.api_key_auth(MY_APP_ID2)
        url = conn.url
        headers['x-api-key'] = conn.api_key
    elif connection_type == ConnectionType.OAUTH2_CLIENT_CREDS:
        # oauth connections contain a url and an access token generated within orchestrate and exposed to your tool
        url = conn.url
        conn = connections.oauth2_client_creds(MY_APP_ID2)
        headers['Authorization'] = f"Bearer {conn.access_token}"
    else:
        raise ValueError(f"Unexpected connection type {connection_type} for {MY_APP_ID2}")

    response = requests.post(
        url,
        headers=headers,
        json={}
    )
    response.raise_for_status()
    return response.json()['result']

​
Importing tools using connections

To import a Python tool that uses connections, use the -a app_id flag to bind each connection to the tool: 
orchestrate tools import -k python \
  -f my_tool.py \
  -a app_1 \
  -a app_2
If your tool expects a specific connection ID in your file, but the actual connection in Orchestrate uses a different name, you can remap it using the -a flag: 
orchestrate tools import -k python \
  -f my_tool.py \
  -a app_id_in_tool=app_id_in_orchestrate

​
Additional functionality of python tools

​
Locally emulating python tool connections

When debugging Python tools outside watsonx Orchestrate, you can run them locally by emulating the environment variables that Orchestrate sets at runtime. These variables expose connection credentials to your tool, allowing you to test functionality without deploying the tool. Orchestrate uses the following naming convention to expose connections: 
export WXO_SECURITY_SCHEMA_<my_app_id>=<connection_type>
export WXO_CONNECTION_<my_app_id>_<field>=<value>
For example, to simulate a connection that uses an API key: 
export WXO_SECURITY_SCHEMA_my_app=api_key_auth
export WXO_CONNECTION_my_app_api_key=your_api_key_here
Set these environment variables before running your script to replicate how Orchestrate injects credentials during execution.
BASH
export WXO_SECURITY_SCHEMA_foo=basic_auth
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_username=username
export WXO_CONNECTION_foo_password=password
BASH
export WXO_SECURITY_SCHEMA_foo=bearer_token
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_token=token
BASH
export WXO_SECURITY_SCHEMA_foo=api_key_auth
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_api_key=apikey
BASH
export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token
BASH
export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token
Coming soon
BASH
export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token
BASH
export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token
BASH
export WXO_SECURITY_SCHEMA_foo=key_value_creds
export WXO_CONNECTION_foo_key1=value1
export WXO_CONNECTION_foo_key2=value2
Note: The app_id must be sanitized by replacing any character that is not alphanumeric (a-z, A-Z, 0-9) with an underscore (_).
sanitize_pattern = re.compile(r"[^a-zA-Z0-9]+")
sanitized_app_id = re.sub(sanitize_pattern,'_', app_id)