> ## Documentation Index
> Fetch the complete documentation index at: https://developer.watson-orchestrate.ibm.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Migrating to Python toolkits

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](/tools/toolkits/choosing_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:**

```python theme={null}
import concurrent.futures
from your_tool import my_tool

def test_concurrent_calls():
    """Test tool with concurrent requests"""
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(my_tool, f"input_{i}") for i in range(10)]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]
    
    # Verify all results are correct and independent
    assert len(results) == 10
    assert len(set(results)) == 10  # All unique if expected

test_concurrent_calls()
```

### 2. Dependency compatibility

Verify that all tools can use the same dependency versions:

```bash theme={null}
# 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:

```bash theme={null}
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:

```bash theme={null}
# Combine all requirements
cat tool1/requirements.txt tool2/requirements.txt tool3/requirements.txt | sort | uniq > my_toolkit/requirements.txt
```

**Resolve version conflicts:**

```txt theme={null}
# 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:**

```txt theme={null}
# ✅ 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):**

```python theme={null}
# 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):**

```python theme={null}
# customer_toolkit/get_customer.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()
```

**Key changes:**

1. Remove global mutable state
2. Ensure all functions are thread-safe and reentrant
3. Use thread-safe libraries

### Step 5: Extract shared utilities

Move common code to shared modules:

**Before (duplicated code):**

```python theme={null}
# tool1.py
def authenticate():
    return {"Authorization": "Bearer token"}

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

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

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

**After (shared utilities):**

```python theme={null}
# 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()
def tool1():
    headers = get_auth_headers()
    # ...

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

@tool()
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:**

```python theme={null}
# Before (standalone)
Tool name: get_customer

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

**Document the name changes:**

```markdown theme={null}
# 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:

```bash theme={null}
# 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:**

```bash theme={null}
# 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):**

```yaml theme={null}
# agent.yaml
spec_version: v1
kind: native
name: support_agent
tools:
  - get_customer
  - update_customer
  - create_ticket
```

**After (toolkit tools):**

```yaml theme={null}
# 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:**

```bash theme={null}
# 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:**

```bash theme={null}
# 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:**

```python theme={null}
# test_concurrency.py
import concurrent.futures
from ibm_watsonx_orchestrate import OrchestrateChatClient

def test_concurrent_requests():
    client = OrchestrateChatClient()
    
    # Send 10 concurrent requests
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(client.chat, agent="support_agent", message=f"Get customer {i}")
            for i in range(10)
        ]
        results = [f.result() for f in concurrent.futures.as_completed(futures)]
    
    # Verify all succeeded
    assert all(r.status == "success" for r in results)
    print("✅ Concurrency test passed")

test_concurrent_requests()
```

**Performance testing:**

```bash theme={null}
# 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:

```bash theme={null}
# 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:

```bash theme={null}
# 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:

1. All agents are updated and tested
2. Toolkit is deployed to live
3. 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:**

1. Create toolkit folder
2. Move tool files
3. Consolidate requirements.txt
4. 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:**

1. Identify duplicated code
2. Extract to shared module
3. Update imports in all tools
4. 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:**

1. Create toolkit folder
2. Move each tool into subfolder
3. Consolidate requirements.txt
4. 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**

```bash theme={null}
# 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**

```python theme={null}
import threading

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

@tool()
def non_threadsafe_tool(input: str) -> str:
    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**

```bash theme={null}
# 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**

```bash theme={null}
# 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:

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

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

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

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

## Validation checklist

Before deploying to production, verify:

* [ ] All tools are thread-safe
* [ ] Dependencies are consolidated and pinned
* [ ] Shared utilities are extracted
* [ ] Tool names are documented
* [ ] Agent configurations are updated
* [ ] Tests pass in draft environment
* [ ] Concurrency tests pass
* [ ] Performance meets expectations
* [ ] Error handling is robust
* [ ] Logging is adequate
* [ ] Documentation is updated
* [ ] Rollback plan is ready

## Rollback procedure

If issues occur after migration:

### Immediate rollback

```bash theme={null}
# 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

```bash theme={null}
# 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

```python theme={null}
# Use connection pooling for HTTP clients
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import threading

# Create a thread-local session for connection reuse
_thread_local = threading.local()

def get_session() -> requests.Session:
    if not hasattr(_thread_local, "session"):
        session = requests.Session()
        adapter = HTTPAdapter(
            pool_connections=10,
            pool_maxsize=20,
            max_retries=Retry(total=3, backoff_factor=0.5)
        )
        session.mount("https://", adapter)
        _thread_local.session = session
    return _thread_local.session

@tool()
def my_tool() -> dict:
    session = get_session()
    response = session.get("https://api.example.com/data", timeout=30)
    return response.json()
```

### 2. Caching

```python theme={null}
from functools import lru_cache

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

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

### 3. Efficient request batching

```python theme={null}
import requests

# ✅ Good: Batch requests where the API supports it
@tool()
def fetch_multiple(ids: list[str]) -> list[dict]:
    response = requests.get(
        "https://api.example.com/items/batch",
        params={"ids": ",".join(ids)},
        timeout=30
    )
    return response.json()

# ✅ Also acceptable: Sequential requests (simple and predictable)
@tool()
def fetch_multiple_sequential(ids: list[str]) -> list[dict]:
    results = []
    session = requests.Session()
    for id in ids:
        response = session.get(f"https://api.example.com/api/{id}", timeout=30)
        results.append(response.json())
    return results
```

## Monitoring and troubleshooting

### Monitor toolkit health

```bash theme={null}
# 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
- 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 threading.Lock for critical sections
- Avoid global mutable state
- Test with high concurrency
```

## Next steps

* [Choosing a tool type](/tools/toolkits/choosing_tool_type)
* [Managing toolkits](/tools/toolkits/manage_toolkits)
* [Python toolkit documentation](/tools/toolkits/python_toolkits)
* [Authoring Python tools](/tools/create_tool)
* [Performance guide](/tutorials/performance/performance-guide-v2-tools)