Skip to main content

Overview

Tools extend agent capabilities by providing access to external systems, APIs, and computational resources. Qwen-Agent includes a rich set of built-in tools and makes it easy to create custom tools.

BaseTool Class

All tools inherit from BaseTool, which defines the tool interface that agents use to interact with them.

Tool Structure

from qwen_agent.tools import BaseTool, register_tool
from typing import Union

@register_tool('my_tool')
class MyTool(BaseTool):
    # Tool metadata
    description = 'A brief description of what this tool does'
    parameters = {
        'type': 'object',
        'properties': {
            'param1': {
                'type': 'string',
                'description': 'Description of param1'
            },
            'param2': {
                'type': 'integer',
                'description': 'Description of param2'
            }
        },
        'required': ['param1']
    }
    
    def call(self, params: Union[str, dict], **kwargs) -> str:
        # Validate and parse parameters
        params_dict = self._verify_json_format_args(params)
        
        # Implement tool logic
        param1 = params_dict['param1']
        param2 = params_dict.get('param2', 0)
        
        result = f"Processed {param1} with {param2}"
        return result
Source Reference: qwen_agent/tools/base.py:109-191

Key Components

name
str
required
Tool identifier. Set via @register_tool('name') decorator or class attribute.
description
str
required
Human-readable description of the tool’s functionality. Used by the LLM to decide when to call the tool.
parameters
Union[List[dict], dict]
required
JSON Schema describing the tool’s parameters. Must follow OpenAI function calling format.

Tool Registration

The @register_tool decorator adds tools to the global registry:
from qwen_agent.tools import TOOL_REGISTRY, register_tool

@register_tool('calculator')
class Calculator(BaseTool):
    description = 'Perform mathematical calculations'
    parameters = {
        'type': 'object',
        'properties': {
            'expression': {
                'type': 'string',
                'description': 'Mathematical expression to evaluate'
            }
        },
        'required': ['expression']
    }
    
    def call(self, params, **kwargs):
        params = self._verify_json_format_args(params)
        result = eval(params['expression'])  # Use safely in production!
        return str(result)

# Tool is now available in TOOL_REGISTRY
print('calculator' in TOOL_REGISTRY)  # True
Source Reference: qwen_agent/tools/base.py:44-59

Built-in Tools

Qwen-Agent provides a comprehensive set of pre-built tools:

Code Execution

CodeInterpreter

Execute Python code in a sandboxed environment.
from qwen_agent.agents import Assistant

agent = Assistant(
    function_list=['code_interpreter'],
    llm={'model': 'qwen-plus'}
)

responses = agent.run_nonstream([
    {'role': 'user', 'content': 'Calculate the first 10 Fibonacci numbers'}
])
Features:
  • Sandboxed execution environment
  • Supports file I/O and plotting
  • Returns text output and generated images
  • Configurable timeout and resource limits
Configuration:
agent = Assistant(
    function_list=[{
        'name': 'code_interpreter',
        'timeout': 30,  # Execution timeout in seconds
        'work_dir': '/path/to/workspace'
    }]
)
Source Reference: qwen_agent/tools/code_interpreter.py:80-100

PythonExecutor

Simpler Python code execution without sandboxing.

Document Processing

DocParser

Parse documents into structured text for processing.
from qwen_agent.tools import DocParser

parser = DocParser({
    'parser_page_size': 500  # Characters per page chunk
})

result = parser.call({
    'url': 'https://example.com/document.pdf'
})
Supported formats: PDF, DOCX, PPTX, TXT, CSV, XLSX, HTML

Retrieval

Retrieve relevant content from documents using RAG.
tool = Retrieval({
    'max_ref_token': 4000,
    'rag_searchers': ['keyword_search', 'vector_search']
})

result = tool.call({
    'query': 'What is machine learning?',
    'files': ['ml_textbook.pdf', 'ai_overview.docx']
})
Search strategies:
  • keyword_search - BM25-based keyword matching
  • vector_search - Semantic similarity search
  • hybrid_search - Combined keyword + vector
  • front_page_search - Search document front matter

Web Tools

WebSearch

Search the web for current information.
agent = Assistant(
    function_list=['web_search'],
    llm={'model': 'qwen-plus'}
)

responses = agent.run_nonstream([
    {'role': 'user', 'content': 'What are the latest developments in AI?'}
])

WebExtractor

Extract and parse content from web pages.

Image Tools

ImageGen

Generate images from text descriptions.
agent = Assistant(
    function_list=['image_gen'],
    llm={'model': 'qwen-plus'}
)

responses = agent.run_nonstream([
    {'role': 'user', 'content': 'Generate an image of a sunset over mountains'}
])

ImageSearch

Search for images based on queries.

Data Management

Storage

Persistent key-value storage for agents.
tool = Storage()

# Store data
tool.call({'key': 'user_pref', 'value': 'dark_mode'})

# Retrieve data
result = tool.call({'key': 'user_pref'})
Source Reference: qwen_agent/tools/__init__.py:15-53

Creating Custom Tools

Simple Tool

import requests
from qwen_agent.tools import BaseTool, register_tool

@register_tool('get_weather')
class WeatherTool(BaseTool):
    description = 'Get current weather for a city'
    parameters = {
        'type': 'object',
        'properties': {
            'city': {
                'type': 'string',
                'description': 'City name, e.g., Beijing, New York'
            },
            'unit': {
                'type': 'string',
                'enum': ['celsius', 'fahrenheit'],
                'description': 'Temperature unit'
            }
        },
        'required': ['city']
    }
    
    def call(self, params, **kwargs):
        params = self._verify_json_format_args(params)
        city = params['city']
        unit = params.get('unit', 'celsius')
        
        # Call weather API
        api_key = 'your-api-key'
        response = requests.get(
            f'https://api.weather.com/v1/current?city={city}&units={unit}',
            headers={'Authorization': f'Bearer {api_key}'}
        )
        
        if response.status_code == 200:
            data = response.json()
            return f"Weather in {city}: {data['temperature']}°, {data['condition']}"
        else:
            return f"Failed to get weather for {city}"

Tool with File Access

For tools that need to work with files:
from qwen_agent.tools import BaseToolWithFileAccess, register_tool
import os

@register_tool('file_analyzer')
class FileAnalyzer(BaseToolWithFileAccess):
    description = 'Analyze files and extract statistics'
    parameters = {
        'type': 'object',
        'properties': {
            'analysis_type': {
                'type': 'string',
                'enum': ['word_count', 'line_count', 'char_count']
            }
        },
        'required': ['analysis_type']
    }
    
    def call(self, params, files=None, **kwargs):
        # files are automatically downloaded to self.work_dir
        params = self._verify_json_format_args(params)
        analysis_type = params['analysis_type']
        
        results = []
        for file_name in os.listdir(self.work_dir):
            file_path = os.path.join(self.work_dir, file_name)
            with open(file_path, 'r') as f:
                content = f.read()
                
            if analysis_type == 'word_count':
                count = len(content.split())
            elif analysis_type == 'line_count':
                count = len(content.split('\n'))
            else:  # char_count
                count = len(content)
            
            results.append(f"{file_name}: {count}")
        
        return '\n'.join(results)
Source Reference: qwen_agent/tools/base.py:193-217

Tool with Configuration

@register_tool('database_query')
class DatabaseQuery(BaseTool):
    description = 'Query a database'
    parameters = {
        'type': 'object',
        'properties': {
            'query': {'type': 'string', 'description': 'SQL query'}
        },
        'required': ['query']
    }
    
    def __init__(self, cfg=None):
        super().__init__(cfg)
        # Access configuration
        self.host = self.cfg.get('host', 'localhost')
        self.port = self.cfg.get('port', 5432)
        self.database = self.cfg.get('database', 'default')
    
    def call(self, params, **kwargs):
        params = self._verify_json_format_args(params)
        query = params['query']
        
        # Connect using self.host, self.port, etc.
        # Execute query
        # Return results
        return "Query results"

# Use with custom configuration
agent = Assistant(
    function_list=[{
        'name': 'database_query',
        'host': 'db.example.com',
        'port': 3306,
        'database': 'production'
    }]
)

Tool Error Handling

from qwen_agent.tools import BaseTool, ToolServiceError, register_tool

@register_tool('api_caller')
class APICaller(BaseTool):
    description = 'Call an external API'
    parameters = {
        'type': 'object',
        'properties': {
            'endpoint': {'type': 'string'}
        },
        'required': ['endpoint']
    }
    
    def call(self, params, **kwargs):
        params = self._verify_json_format_args(params)
        
        try:
            response = requests.get(params['endpoint'], timeout=10)
            
            if response.status_code == 429:
                # Raise ToolServiceError for rate limiting
                raise ToolServiceError(
                    code='429',
                    message='Rate limit exceeded. Please try again later.',
                    extra={'retry_after': response.headers.get('Retry-After')}
                )
            
            response.raise_for_status()
            return response.text
            
        except requests.RequestException as e:
            # The agent will catch and log this
            raise ToolServiceError(
                exception=e,
                code='500',
                message=f'API request failed: {str(e)}'
            )
Source Reference: qwen_agent/tools/base.py:27-41

Using Tools in Agents

String Names

from qwen_agent.agents import Assistant

agent = Assistant(
    function_list=['code_interpreter', 'image_gen', 'web_search']
)

Configured Tools

agent = Assistant(
    function_list=[
        {'name': 'code_interpreter', 'timeout': 60},
        {'name': 'retrieval', 'max_ref_token': 8000},
    ]
)

Tool Instances

from qwen_agent.tools import CodeInterpreter, ImageGen

code_tool = CodeInterpreter({'timeout': 30})
image_tool = ImageGen()

agent = Assistant(
    function_list=[code_tool, image_tool]
)

Custom Tools

from my_tools import WeatherTool, DatabaseQuery

# Registered tools can be used by name
agent = Assistant(
    function_list=['get_weather', 'database_query']
)

# Or pass instances with config
agent = Assistant(
    function_list=[
        WeatherTool(),
        DatabaseQuery({'host': 'db.example.com'})
    ]
)

MCP Integration

Qwen-Agent supports Model Context Protocol (MCP) servers for dynamic tool loading:
agent = Assistant(
    function_list=[{
        'mcpServers': {
            'github': {
                'command': 'npx',
                'args': ['-y', '@modelcontextprotocol/server-github']
            }
        }
    }]
)

Parameter Schema Format

Tools use OpenAI-compatible JSON Schema for parameters:
# Simple schema
parameters = {
    'type': 'object',
    'properties': {
        'name': {
            'type': 'string',
            'description': 'User name'
        },
        'age': {
            'type': 'integer',
            'description': 'User age',
            'minimum': 0,
            'maximum': 120
        },
        'category': {
            'type': 'string',
            'enum': ['A', 'B', 'C'],
            'description': 'Category selection'
        }
    },
    'required': ['name']
}

# Nested objects
parameters = {
    'type': 'object',
    'properties': {
        'user': {
            'type': 'object',
            'properties': {
                'name': {'type': 'string'},
                'email': {'type': 'string'}
            },
            'required': ['name', 'email']
        },
        'tags': {
            'type': 'array',
            'items': {'type': 'string'}
        }
    },
    'required': ['user']
}
Source Reference: qwen_agent/tools/base.py:62-106

Best Practices

Clear Descriptions

  • Write descriptive tool and parameter descriptions
  • The LLM uses these to decide when to call tools
  • Be specific about input formats and expectations
  • Include examples in descriptions when helpful

Robust Error Handling

  • Always validate inputs with _verify_json_format_args()
  • Use ToolServiceError for expected errors
  • Let unexpected exceptions propagate (agent will handle)
  • Provide helpful error messages

Efficient Execution

  • Set reasonable timeouts for long-running operations
  • Use async operations when possible
  • Cache results when appropriate
  • Clean up resources in tool implementation

Schema Design

  • Use enums for constrained choices
  • Mark truly required fields as required
  • Provide sensible defaults
  • Keep parameter lists focused and minimal

Testing Tools

import pytest
from my_tools import WeatherTool

def test_weather_tool():
    tool = WeatherTool()
    
    # Test valid input
    result = tool.call({'city': 'Beijing', 'unit': 'celsius'})
    assert 'Weather in Beijing' in result
    
    # Test default parameter
    result = tool.call({'city': 'Tokyo'})
    assert result is not None
    
    # Test invalid input
    with pytest.raises(ValueError):
        tool.call({'unit': 'celsius'})  # Missing required 'city'
    
    # Test parameter validation
    params = tool._verify_json_format_args('{"city": "Paris"}')
    assert params['city'] == 'Paris'

Function Calling

Learn how agents use function calling to invoke tools

Agents

Understand how agents integrate with tools

Build docs developers (and LLMs) love