Associating connections to Python tools - IBM watsonx Orchestrate ADK

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:

Basic Connections

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']

Bearer Connections

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']

API Key Connections

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']

OAuth (Client Credentials)

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']

OAuth (Auth Code)

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']

🚧 OAuth (Implicit)

Coming soon

OAuth (Password)

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']

🌐OAuth (SSO/IDP Flow)

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']

Key Value Connections

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.

Basic Connections

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

Bearer Connections

BASH

export WXO_SECURITY_SCHEMA_foo=bearer_token
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_token=token

API Key Connections

BASH

export WXO_SECURITY_SCHEMA_foo=api_key_auth
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_api_key=apikey

OAuth (Client Credentials)

BASH

export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token

OAuth (Auth Code)

BASH

export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token

🚧 OAuth (Implicit)

Coming soon

OAuth (Password)

BASH

export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token

🌐 OAuth (SSO/IDP Flow)

BASH

export WXO_SECURITY_SCHEMA_foo=oauth2
export WXO_CONNECTION_foo_url=apikey
export WXO_CONNECTION_foo_access_token=token

Key Value Connections

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)

​Defining a python tools expected credentials

​Examples of fetching credentials per connection type

​Combining multiple connection types for a single application id

​Importing tools using connections

​Additional functionality of python tools

​Locally emulating python tool connections

Defining a python tools expected credentials

Examples of fetching credentials per connection type

Combining multiple connection types for a single application id

Importing tools using connections

Additional functionality of python tools

Locally emulating python tool connections