Migrating to Python toolkits - IBM watsonx Orchestrate ADK

This guide helps you migrate existing standalone Python tools into Python toolkits for improved performance and resource efficiency in production environments.

Why migrate to Python toolkits?

Python toolkits offer significant advantages over standalone Python tools in production: Performance benefits:

No process overhead: Tools run in a persistent process, eliminating 100-300ms startup time per call
Shared resources: Tools share dependencies and memory, reducing overall resource usage
Dedicated containers: In live premium environments, each toolkit gets its own container with 2 vCPUs and 2 GB memory

Operational benefits:

Atomic updates: All tools in a toolkit update together, ensuring consistency
Simplified management: Deploy and manage related tools as a single unit
Better scaling: Dedicated resources in live environments support higher concurrency

When to migrate:

Tools are called frequently (>10 times per minute)
You have multiple related tools that share dependencies
You need better performance in production
You’ve verified all tools are thread-safe

When NOT to migrate:

Tools use non-thread-safe operations
Tools are called infrequently
Tools have conflicting dependencies
You need independent tool updates

For help deciding, see Choosing a tool type.

Prerequisites

Before migrating, ensure your tools meet these requirements:

1. Thread-safety verification

All tools in a Python toolkit must be thread-safe. Review each tool for: Common thread-safety issues:

Global mutable state
File system writes
Non-thread-safe libraries
Shared database connections without pooling
Race conditions in shared resources

Testing for thread-safety:

import asyncio
import concurrent.futures
from your_tool import my_tool

async def test_concurrent_calls():
    """Test tool with concurrent requests"""
    tasks = [my_tool(f"input_{i}") for i in range(10)]
    results = await asyncio.gather(*tasks)
    
    # Verify all results are correct and independent
    assert len(results) == 10
    assert len(set(results)) == 10  # All unique if expected

# Run the test
asyncio.run(test_concurrent_calls())

2. Dependency compatibility

Verify that all tools can use the same dependency versions:

# Check for conflicting dependencies
cat tool1/requirements.txt tool2/requirements.txt | sort | uniq -d

If tools require different versions of the same package, they cannot be in the same toolkit.

3. Shared utility functions

Identify common code that can be shared:

Authentication helpers
Data transformation utilities
API client wrappers
Validation functions

Migration process

Step 1: Identify candidate tools

List all standalone Python tools and group them by: Functional relationship:

Tools that work with the same API or service
Tools that perform related operations
Tools that share business logic

Usage patterns:

Frequently called tools (>10 calls/minute)
Tools called together by the same agents
Tools with similar performance requirements

Example grouping:

Customer Service Toolkit:
  - get_customer_info
  - update_customer_address
  - create_support_ticket

Data Processing Toolkit:
  - validate_csv
  - transform_data
  - export_results

Step 2: Create toolkit folder structure

Organize your tools into a toolkit folder:

my_toolkit/
├── requirements.txt          # Shared dependencies
├── __init__.py              # Optional: shared utilities
├── tool1.py                 # First tool
├── tool2.py                 # Second tool
├── tool3.py                 # Third tool
└── utils/                   # Optional: shared modules
    ├── __init__.py
    ├── auth.py
    └── helpers.py

Best practices:

Use descriptive folder names (e.g., customer_service_toolkit)
Keep one tool per file for clarity
Place shared code in utils/ or __init__.py
Include comprehensive requirements.txt

Step 3: Consolidate dependencies

Merge requirements.txt files from all tools:

# Combine all requirements
cat tool1/requirements.txt tool2/requirements.txt tool3/requirements.txt | sort | uniq > my_toolkit/requirements.txt

Resolve version conflicts:

# Before (conflicting versions)
# tool1/requirements.txt
requests==2.28.0

# tool2/requirements.txt
requests==2.31.0

# After (choose compatible version)
# my_toolkit/requirements.txt
requests==2.31.0  # Use latest compatible version

Pin all dependencies:

# ✅ Good: Exact versions
requests==2.31.0
pydantic==2.5.0
httpx==0.25.0

# ❌ Bad: Unpinned versions
requests
pydantic>=2.0
httpx~=0.25

Step 4: Refactor tool code

Update each tool file to work in a toolkit context: Before (standalone tool):

# customer_tool.py
from ibm_watsonx_orchestrate.agent_builder.tools import tool
import requests

@tool(name="get_customer")
def get_customer(customer_id: str) -> dict:
    """Get customer information"""
    response = requests.get(f"https://api.example.com/customers/{customer_id}")
    return response.json()

After (toolkit tool):

# customer_toolkit/get_customer.py
from ibm_watsonx_orchestrate.agent_builder.tools import tool
import httpx  # Use async HTTP client

@tool(name="get_customer")
async def get_customer(customer_id: str) -> dict:
    """Get customer information"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/customers/{customer_id}")
        return response.json()

Key changes:

Use async def for all tool functions
Use async libraries (httpx instead of requests)
Remove global mutable state
Use proper async patterns

Step 5: Extract shared utilities

Move common code to shared modules: Before (duplicated code):

# tool1.py
def authenticate():
    return {"Authorization": "Bearer token"}

@tool()
async def tool1():
    headers = authenticate()
    # ...

# tool2.py
def authenticate():
    return {"Authorization": "Bearer token"}

@tool()
async def tool2():
    headers = authenticate()
    # ...

After (shared utilities):

# my_toolkit/utils/auth.py
def get_auth_headers() -> dict:
    """Get authentication headers"""
    return {"Authorization": "Bearer token"}

# my_toolkit/tool1.py
from .utils.auth import get_auth_headers

@tool()
async def tool1():
    headers = get_auth_headers()
    # ...

# my_toolkit/tool2.py
from .utils.auth import get_auth_headers

@tool()
async def tool2():
    headers = get_auth_headers()
    # ...

Step 6: Update tool names

When you import a toolkit, tool names change to include the toolkit prefix: Naming convention:

toolkit_name:tool_name

Example:

# Before (standalone)
Tool name: get_customer

# After (in toolkit named "customer_service")
Tool name: customer_service:get_customer

Document the name changes:

# Tool Name Mapping

| Old Name | New Name |
|----------|----------|
| get_customer | customer_service:get_customer |
| update_customer | customer_service:update_customer |
| create_ticket | customer_service:create_ticket |

Step 7: Import the toolkit

Import your toolkit using the ADK CLI:

# Import toolkit
orchestrate toolkits add \
  --kind python \
  --name customer_service \
  --description "Customer service tools for support agents" \
  --package_root ./customer_service_toolkit \
  --tier small

Verify import:

# List all tools in the toolkit
orchestrate tools list | grep "customer_service:"

Expected output:

customer_service:get_customer
customer_service:update_customer
customer_service:create_ticket

Step 8: Update agent configurations

Update all agents that use the migrated tools: Before (standalone tools):

# agent.yaml
spec_version: v1
kind: native
name: support_agent
tools:
  - get_customer
  - update_customer
  - create_ticket

After (toolkit tools):

# agent.yaml
spec_version: v1
kind: native
name: support_agent
tools:
  - customer_service:get_customer
  - customer_service:update_customer
  - customer_service:create_ticket

Update and redeploy agents:

# Update agent configuration
orchestrate agents import -f agent.yaml

# Deploy to live environment
orchestrate agents publish -n support_agent

Step 9: Test in draft environment

Thoroughly test the toolkit before deploying to live: Functional testing:

# Test each tool individually
orchestrate agents chat -n support_agent -m "Get customer info for ID 12345"
orchestrate agents chat -n support_agent -m "Update customer address"
orchestrate agents chat -n support_agent -m "Create a support ticket"

Concurrency testing:

# test_concurrency.py
import asyncio
from ibm_watsonx_orchestrate import OrchestrateChatClient

async def test_concurrent_requests():
    client = OrchestrateChatClient()
    
    # Send 10 concurrent requests
    tasks = [
        client.chat(agent="support_agent", message=f"Get customer {i}")
        for i in range(10)
    ]
    
    results = await asyncio.gather(*tasks)
    
    # Verify all succeeded
    assert all(r.status == "success" for r in results)
    print("✅ Concurrency test passed")

asyncio.run(test_concurrent_requests())

Performance testing:

# Measure response times
time orchestrate agents chat -n support_agent -m "Get customer 12345"

Compare with standalone tool performance to verify improvement.

Step 10: Deploy to live environment

After successful testing in draft:

# Publish agent to live
orchestrate agents publish -n support_agent

# Verify in live environment
orchestrate env switch live
orchestrate tools list | grep "customer_service:"

Monitor performance:

Check response times
Monitor error rates
Verify concurrent request handling
Review resource usage

Step 11: Remove old standalone tools

After confirming the toolkit works correctly:

# Remove old standalone tools
orchestrate tools remove -n get_customer
orchestrate tools remove -n update_customer
orchestrate tools remove -n create_ticket

Important: Only remove standalone tools after:

All agents are updated and tested
Toolkit is deployed to live
No agents reference the old tool names

Common migration patterns

Pattern 1: Simple consolidation

Scenario: Multiple independent tools with no shared code Before:

tools/
├── tool1.py
├── tool2.py
└── tool3.py

After:

my_toolkit/
├── requirements.txt
├── tool1.py
├── tool2.py
└── tool3.py

Steps:

Create toolkit folder
Move tool files
Consolidate requirements.txt
Import as toolkit

Pattern 2: Shared utilities extraction

Scenario: Tools with duplicated helper functions Before:

tools/
├── tool1.py  # Contains auth_helper()
├── tool2.py  # Contains auth_helper() (duplicate)
└── tool3.py  # Contains auth_helper() (duplicate)

After:

my_toolkit/
├── requirements.txt
├── __init__.py
├── utils/
│   ├── __init__.py
│   └── auth.py  # Shared auth_helper()
├── tool1.py     # Imports from utils.auth
├── tool2.py     # Imports from utils.auth
└── tool3.py     # Imports from utils.auth

Steps:

Identify duplicated code
Extract to shared module
Update imports in all tools
Test thoroughly

Pattern 3: Multi-file tool consolidation

Scenario: Standalone tools with package roots Before:

tool1/
├── requirements.txt
├── main.py
└── helpers/
    └── utils.py

tool2/
├── requirements.txt
├── main.py
└── helpers/
    └── utils.py

After:

my_toolkit/
├── requirements.txt
├── tool1/
│   ├── main.py
│   └── helpers/
│       └── utils.py
└── tool2/
    ├── main.py
    └── helpers/
        └── utils.py

Steps:

Create toolkit folder
Move each tool into subfolder
Consolidate requirements.txt
Update relative imports if needed

Handling special cases

Non-thread-safe tools

If a tool cannot be made thread-safe: Option 1: Keep as standalone tool

# Keep problematic tool separate
orchestrate tools import -k python -f non_threadsafe_tool.py

# Import other tools as toolkit
orchestrate toolkits add --kind python --name my_toolkit --package_root ./toolkit

Option 2: Add synchronization

import asyncio

# Add lock for non-thread-safe operations
_lock = asyncio.Lock()

@tool()
async def non_threadsafe_tool(input: str) -> str:
    async with _lock:
        # Critical section - only one thread at a time
        result = perform_non_threadsafe_operation(input)
    return result

Conflicting dependencies

If tools require different versions: Option 1: Separate toolkits

# Create separate toolkits
orchestrate toolkits add --kind python --name toolkit_a --package_root ./toolkit_a
orchestrate toolkits add --kind python --name toolkit_b --package_root ./toolkit_b

Option 2: Update to compatible versions

# Test with newer version
pip install package==2.0.0
# Run tests
pytest tests/

# If successful, update requirements.txt
echo "package==2.0.0" >> requirements.txt

Connection remapping

If tools use different connection names:

# Before (tool1 uses "api_connection")
@tool(expected_credentials=[{"app_id": "api_connection", "type": ConnectionType.API_KEY}])
async def tool1():
    conn = connections.api_key("api_connection")

# Before (tool2 uses "service_connection")
@tool(expected_credentials=[{"app_id": "service_connection", "type": ConnectionType.API_KEY}])
async def tool2():
    conn = connections.api_key("service_connection")

# After (standardize on one connection)
@tool(expected_credentials=[{"app_id": "shared_connection", "type": ConnectionType.API_KEY}])
async def tool1():
    conn = connections.api_key("shared_connection")

@tool(expected_credentials=[{"app_id": "shared_connection", "type": ConnectionType.API_KEY}])
async def tool2():
    conn = connections.api_key("shared_connection")

Validation checklist

Before deploying to production, verify:

Rollback procedure

If issues occur after migration:

Immediate rollback

# Switch back to draft
orchestrate env switch draft

# Reimport standalone tools
orchestrate tools import -k python -f tool1.py
orchestrate tools import -k python -f tool2.py
orchestrate tools import -k python -f tool3.py

# Revert agent configurations
orchestrate agents import -f agent_backup.yaml

# Publish to live
orchestrate agents publish -n support_agent

Gradual rollback

# Keep toolkit but add standalone versions
orchestrate tools import -k python -f tool1.py -n tool1_standalone

# Update agent to use standalone version
# Edit agent.yaml to use tool1_standalone instead of toolkit:tool1

# Test and verify
orchestrate agents chat -n support_agent -m "test message"

# If successful, remove toolkit
orchestrate toolkits remove -n my_toolkit

Performance optimization

After migration, optimize toolkit performance:

1. Connection pooling

# Use connection pooling for HTTP clients
import httpx

# Create shared client (reused across requests)
_http_client = None

async def get_http_client():
    global _http_client
    if _http_client is None:
        _http_client = httpx.AsyncClient(
            timeout=30.0,
            limits=httpx.Limits(max_connections=100)
        )
    return _http_client

@tool()
async def my_tool():
    client = await get_http_client()
    response = await client.get("https://api.example.com/data")
    return response.json()

2. Caching

from functools import lru_cache
import asyncio

# Cache expensive computations
@lru_cache(maxsize=128)
def expensive_computation(input: str) -> str:
    # Expensive operation
    return result

@tool()
async def my_tool(input: str) -> str:
    result = expensive_computation(input)
    return result

3. Async patterns

# ✅ Good: Concurrent operations
@tool()
async def fetch_multiple(ids: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [client.get(f"/api/{id}") for id in ids]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

# ❌ Bad: Sequential operations
@tool()
async def fetch_multiple(ids: list[str]) -> list[dict]:
    results = []
    async with httpx.AsyncClient() as client:
        for id in ids:
            response = await client.get(f"/api/{id}")
            results.append(response.json())
    return results

Monitoring and troubleshooting

Monitor toolkit health

# Check toolkit status
orchestrate toolkits list -v

# View tool execution logs
orchestrate logs --agent support_agent --filter "customer_service:"

# Monitor resource usage
orchestrate toolkits stats -n customer_service

Common issues

Issue: Tools timing out

Solution: Increase timeout or optimize tool code
- Check for blocking operations
- Use async libraries
- Add connection pooling

Issue: Memory errors

Solution: Reduce memory usage
- Limit concurrent requests
- Clear caches periodically
- Process data in chunks

Issue: Race conditions

Solution: Add synchronization
- Use asyncio.Lock for critical sections
- Avoid global mutable state
- Test with high concurrency

​Why migrate to Python toolkits?

​Prerequisites

​1. Thread-safety verification

​2. Dependency compatibility

​3. Shared utility functions

​Migration process

​Step 1: Identify candidate tools

​Step 2: Create toolkit folder structure

​Step 3: Consolidate dependencies

​Step 4: Refactor tool code

​Step 5: Extract shared utilities

​Step 6: Update tool names

​Step 7: Import the toolkit

​Step 8: Update agent configurations

​Step 9: Test in draft environment

​Step 10: Deploy to live environment

​Step 11: Remove old standalone tools

​Common migration patterns

​Pattern 1: Simple consolidation

​Pattern 2: Shared utilities extraction

​Pattern 3: Multi-file tool consolidation

​Handling special cases

​Non-thread-safe tools

​Conflicting dependencies

​Connection remapping

​Validation checklist

​Rollback procedure

​Immediate rollback

​Gradual rollback

​Performance optimization

​1. Connection pooling

​2. Caching

​3. Async patterns

​Monitoring and troubleshooting

​Monitor toolkit health

​Common issues

​Next steps

Why migrate to Python toolkits?

Prerequisites

1. Thread-safety verification

2. Dependency compatibility

3. Shared utility functions

Migration process

Step 1: Identify candidate tools

Step 2: Create toolkit folder structure

Step 3: Consolidate dependencies

Step 4: Refactor tool code

Step 5: Extract shared utilities

Step 6: Update tool names

Step 7: Import the toolkit

Step 8: Update agent configurations

Step 9: Test in draft environment

Step 10: Deploy to live environment

Step 11: Remove old standalone tools

Common migration patterns

Pattern 1: Simple consolidation

Pattern 2: Shared utilities extraction

Pattern 3: Multi-file tool consolidation

Handling special cases

Non-thread-safe tools

Conflicting dependencies

Connection remapping

Validation checklist

Rollback procedure

Immediate rollback

Gradual rollback

Performance optimization

1. Connection pooling

2. Caching

3. Async patterns

Monitoring and troubleshooting

Monitor toolkit health

Common issues

Next steps