Tool Versioning and Compatibility

Why Tool Versioning Matters

Semantic Versioning for Tools

VERSIONING_RULES = {
    'major_bump': [
        'Removed a required or optional parameter',
        'Renamed an existing parameter',
        'Changed parameter type (e.g., string -> object)',
        'Changed response field names or types',
        'Removed a response field',
        'Changed error code values'
    ],
    'minor_bump': [
        'Added an optional parameter',
        'Added a new response field',
        'Added a new tool to the plugin'
    ],
    'patch_bump': [
        'Fixed a bug without interface change',
        'Improved error messages',
        'Performance improvement',
        'Updated documentation'
    ]
}

for bump_type, examples in VERSIONING_RULES.items():
    print(f'{bump_type}:')
    for ex in examples[:2]:
        print(f'  - {ex}')

Parsing and Comparing Versions

from functools import total_ordering

@total_ordering
class Version:
    def __init__(self, version_str: str):
        parts = version_str.strip().split('.')
        if len(parts) != 3 or not all(p.isdigit() for p in parts):
            raise ValueError(f'Invalid version: {version_str}')
        self.major, self.minor, self.patch = map(int, parts)

    def __str__(self):
        return f'{self.major}.{self.minor}.{self.patch}'

    def __eq__(self, other):
        return (self.major, self.minor, self.patch) == (other.major, other.minor, other.patch)

    def __lt__(self, other):
        return (self.major, self.minor, self.patch) < (other.major, other.minor, other.patch)

    def is_compatible_with(self, required: 'Version') -> bool:
        """Compatible if same major version and >= required minor.patch"""
        return self.major == required.major and self >= required

v = Version('2.3.1')
required = Version('2.1.0')
print(f'{v} compatible with {required}: {v.is_compatible_with(required)}')
print(f'Is newer: {v > required}')

Version Pinning in Agent Config

AGENT_TOOL_REQUIREMENTS = {
    'get_weather': '>=1.2.0',
    'search_knowledge_base': '>=3.0.0',
    'send_email': '>=2.1.0,<3.0.0'  # exclude major-version bump
}

def parse_version_constraint(constraint: str) -> list:
    """
    Parses constraints like '>=1.2.0,<3.0.0'
    Returns list of (operator, Version) tuples
    """
    ops = {'>=': lambda a, b: a >= b, '>': lambda a, b: a > b,
           '<=': lambda a, b: a <= b, '<': lambda a, b: a < b,
           '==': lambda a, b: a == b}
    parts = [p.strip() for p in constraint.split(',')]
    parsed = []
    for part in parts:
        for op_str, op_fn in ops.items():
            if part.startswith(op_str):
                parsed.append((op_fn, Version(part[len(op_str):])))
                break
    return parsed

def satisfies_constraint(tool_version: str, constraint: str) -> bool:
    v = Version(tool_version)
    rules = parse_version_constraint(constraint)
    return all(op(v, required) for op, required in rules)

print(satisfies_constraint('2.3.0', '>=2.1.0,<3.0.0'))  # True
print(satisfies_constraint('3.0.0', '>=2.1.0,<3.0.0'))  # False

Deprecation Warnings

import warnings

def execute_get_weather(params: dict) -> dict:
    # Handle deprecated parameter 'temp_unit' -> replaced by 'units'
    if 'temp_unit' in params:
        warnings.warn(
            'Parameter temp_unit is deprecated since v1.3.0. '
            'Use units instead. Will be removed in v2.0.0.',
            DeprecationWarning,
            stacklevel=2
        )
        params = dict(params)
        params['units'] = params.pop('temp_unit')

    # Include deprecation notice in response
    result = _fetch_weather(params['city'], params.get('units', 'celsius'))
    if 'temp_unit' in params:
        result['_deprecation_warnings'] = [
            'temp_unit deprecated; use units'
        ]
    return result

# When calling deprecated param:
with warnings.catch_warnings(record=True) as w:
    warnings.simplefilter('always')
    # result = execute_get_weather({'city': 'London', 'temp_unit': 'celsius'})
    print('Deprecation warnings would be captured here')

Migration Guides

MIGRATION_GUIDES = {
    '1.x_to_2.0': {
        'summary': 'Response structure changed: temperature is now nested under data{}',
        'breaking_changes': [
            {
                'description': 'temperature field moved',
                'before': 'result["temperature"]',
                'after': 'result["data"]["temperature"]'
            },
            {
                'description': 'temp_unit parameter removed',
                'before': 'execute({"city": "London", "temp_unit": "celsius"})',
                'after': 'execute({"city": "London", "units": "celsius"})'
            }
        ],
        'migration_steps': [
            '1. Update parameter name: temp_unit -> units',
            '2. Update response access: result["temperature"] -> result["data"]["temperature"]',
            '3. Run your test suite against v2.0.0'
        ]
    }
}

for guide_key, guide in MIGRATION_GUIDES.items():
    print(f'Migration guide {guide_key}:')
    print(f'  {guide["summary"]}')
    print(f'  Steps: {len(guide["migration_steps"])}')

Compatibility Matrix

COMPATIBILITY_MATRIX = {
    'weather-tools': {
        '1.x': {'framework_min': '0.8.0', 'framework_max': '0.x.x', 'status': 'EOL'},
        '2.x': {'framework_min': '1.0.0', 'framework_max': '1.x.x', 'status': 'supported'},
        '3.x': {'framework_min': '2.0.0', 'framework_max': None, 'status': 'latest'}
    }
}

def check_compatibility(
    plugin_name: str,
    tool_version: str,
    framework_version: str
) -> dict:
    matrix = COMPATIBILITY_MATRIX.get(plugin_name, {})
    major = tool_version.split('.')[0] + '.x'
    row = matrix.get(major)
    if not row:
        return {'compatible': False, 'reason': 'Version not in matrix'}
    fw = Version(framework_version)
    min_fw = Version(row['framework_min'])
    compatible = fw >= min_fw
    return {'compatible': compatible, 'status': row['status'],
            'min_framework': row['framework_min']}

result = check_compatibility('weather-tools', '2.3.0', '1.2.0')
print(result)

Version Registry

class VersionedToolRegistry(ToolRegistry):
    def register_tool(self, name, schema, execute_fn, plugin_name):
        if name in self._tools:
            existing_v = Version(self._tools[name]['schema'].get('version', '0.0.0'))
            new_v = Version(schema.get('version', '0.0.0'))
            if new_v > existing_v:
                print(f'Upgrading tool {name}: {existing_v} -> {new_v}')
            else:
                print(f'Keeping tool {name} v{existing_v} '
                      f'(skipping older v{new_v} from {plugin_name})')
                return
        super().register_tool(name, schema, execute_fn, plugin_name)

    def get_version(self, tool_name: str) -> str:
        tool = self._tools.get(tool_name)
        if not tool:
            return None
        return tool['schema'].get('version', 'unknown')

    def check_requirement(self, tool_name: str, constraint: str) -> bool:
        v = self.get_version(tool_name)
        if not v:
            return False
        return satisfies_constraint(v, constraint)

Handling Version Mismatches Gracefully

class VersionCheckResult:
    def __init__(self):
        self.satisfied = []
        self.unsatisfied = []
        self.missing = []

def check_all_requirements(
    requirements: dict,
    registry
) -> VersionCheckResult:
    result = VersionCheckResult()
    for tool_name, constraint in requirements.items():
        installed_v = registry.get_version(tool_name)
        if installed_v is None:
            result.missing.append(tool_name)
        elif not satisfies_constraint(installed_v, constraint):
            result.unsatisfied.append({
                'tool': tool_name,
                'required': constraint,
                'installed': installed_v
            })
        else:
            result.satisfied.append(tool_name)
    return result

def start_agent_with_version_check(requirements, registry):
    check = check_all_requirements(requirements, registry)
    if check.missing:
        raise RuntimeError(f'Missing tools: {check.missing}')
    if check.unsatisfied:
        for item in check.unsatisfied:
            print(f'VERSION MISMATCH: {item["tool"]} '
                  f'requires {item["required"]}, got {item["installed"]}')
        raise RuntimeError('Tool version requirements not satisfied')
    print('All tool requirements satisfied')

Changelog Automation

# Conventional commit format:
# feat!: (major) remove temp_unit parameter
# feat: (minor) add 'humidity_pct' to response
# fix: (patch) handle API timeout correctly
# docs: update README

import subprocess

def generate_changelog_from_git(
    from_tag: str = 'v1.2.0',
    to_tag: str = 'HEAD'
) -> dict:
    try:
        log = subprocess.check_output(
            ['git', 'log', f'{from_tag}..{to_tag}',
             '--oneline', '--pretty=format:%s'],
            text=True
        ).strip().split('\n')
    except subprocess.CalledProcessError:
        return {'error': 'git log failed'}

    changelog = {'breaking': [], 'features': [], 'fixes': [], 'docs': []}
    for msg in log:
        if msg.startswith('feat!'):
            changelog['breaking'].append(msg[5:].strip())
        elif msg.startswith('feat:'):
            changelog['features'].append(msg[5:].strip())
        elif msg.startswith('fix:'):
            changelog['fixes'].append(msg[4:].strip())
        elif msg.startswith('docs:'):
            changelog['docs'].append(msg[5:].strip())
    return changelog

if __name__ == '__main__':
    subprocess.check_output = lambda *a, **k: (
        "feat!: remove temp_unit parameter\n"
        "feat: add humidity_pct to response\n"
        "fix: handle API timeout correctly\n"
        "docs: update README"
    )
    changelog = generate_changelog_from_git()
    print('Changelog:')
    for section, items in changelog.items():
        print(f'  {section}: {items}')

Lock Files for Reproducibility

import json
import os

LOCK_FILE = '.tool-lock.json'

def generate_lock_file(registry) -> dict:
    lock = {
        'generated_at': __import__('datetime').datetime.utcnow().isoformat(),
        'tools': {}
    }
    for tool_name, tool_info in registry._tools.items():
        lock['tools'][tool_name] = {
            'version': tool_info['schema'].get('version', 'unknown'),
            'plugin': tool_info['plugin']
        }
    with open(LOCK_FILE, 'w') as f:
        json.dump(lock, f, indent=2)
    print(f'Lock file written: {len(lock["tools"])} tools')
    return lock

def verify_lock_file(registry) -> bool:
    if not os.path.exists(LOCK_FILE):
        print('No lock file found — run generate_lock_file() first')
        return False
    with open(LOCK_FILE) as f:
        lock = json.load(f)
    for tool_name, locked_info in lock['tools'].items():
        installed_v = registry.get_version(tool_name)
        if installed_v != locked_info['version']:
            print(f'VERSION MISMATCH: {tool_name} locked={locked_info["version"]} installed={installed_v}')
            return False
    print('Lock file verified: all versions match')
    return True

if __name__ == '__main__':
    import tempfile
    os.chdir(tempfile.gettempdir())

    class MockRegistry:
        def __init__(self):
            self._tools = {'get_weather': {'schema': {'version': '1.2.0'}, 'plugin': 'weather-tools'}}
        def get_version(self, name):
            return self._tools[name]['schema']['version']

    registry = MockRegistry()
    generate_lock_file(registry)
    verify_lock_file(registry)

Knowledge Check

Recap: Tool Versioning and Compatibility