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.
All tools inherit from BaseTool, which defines the tool interface that agents use to interact with them.
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
Tool identifier. Set via @register_tool('name') decorator or class attribute.
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.
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
Qwen-Agent provides a comprehensive set of pre-built tools:
Code Execution
CodeInterpreter
Execute Python code in a sandboxed environment.
Using CodeInterpreter
Direct Usage
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
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?' }
])
Extract and parse content from web pages.
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
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 } "
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
@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'
}]
)
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
String Names
from qwen_agent.agents import Assistant
agent = Assistant(
function_list = [ 'code_interpreter' , 'image_gen' , 'web_search' ]
)
agent = Assistant(
function_list = [
{ 'name' : 'code_interpreter' , 'timeout' : 60 },
{ 'name' : 'retrieval' , 'max_ref_token' : 8000 },
]
)
from qwen_agent.tools import CodeInterpreter, ImageGen
code_tool = CodeInterpreter({ 'timeout' : 30 })
image_tool = ImageGen()
agent = Assistant(
function_list = [code_tool, image_tool]
)
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' ]
}
}
}]
)
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
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