Introduzione: Il Tool Calling come Ponte verso il Mondo Reale
Il tool calling è il meccanismo che permette agli agenti AI di superare il confine della generazione di testo e agire nel mondo reale. Senza tool calling, un agente può solo produrre parole: risposte, spiegazioni, codice che resta sullo schermo. Con il tool calling, può cercare informazioni sul web, interrogare database, chiamare API esterne, creare file, inviare email, gestire deploy e automatizzare processi complessi.
Il tool calling trasforma il modello linguistico da un sistema di generazione passiva a un orchestratore attivo che decide autonomamente quali strumenti usare, con quali parametri invocarli e come comporre i risultati in una risposta coerente. Questa capacità e ciò che distingue un chatbot da un agente: il primo risponde, il secondo agisce.
In questo articolo esploreremo il tool calling in profondità: dalla definizione formale di un tool con JSON Schema, alla validazione degli input, al parsing degli output, fino all'integrazione con API REST, GraphQL e database. Costruiremo un framework di tool riutilizzabili e analizzeremo pattern avanzati come il tool discovery dinamico e la gestione di tool long-running.
Cosa Imparerai in Questo Articolo
- Come definire tool con JSON Schema: nome, descrizione, parametri e output
- Input validation e sanitization per prevenire injection e errori
- Output parsing strutturato con error recovery
- Integrazione con API REST e auto-generazione da OpenAPI spec
- Database tools con query sicure e parameterizzate
- Come costruire un custom tool framework riutilizzabile
- Tool discovery dinamico e registration a runtime
- Gestione di tool long-running con streaming e timeout
Tool Specification: JSON Schema
Ogni tool che un agente può utilizzare deve essere descritto formalmente in modo che il modello linguistico comprenda cosa fa, quali parametri accetta e che tipo di output produce. Lo standard de facto per questa descrizione e JSON Schema, un formato dichiarativo che permette di specificare la struttura, i tipi e i vincoli dei dati.
Una buona tool specification è fondamentale per la qualità dell'agente: se la descrizione è vaga, il modello non sapra quando usare il tool; se i parametri sono ambigui, generera chiamate con valori errati; se l'output non è documentato, non sarà in grado di interpretare correttamente i risultati.
Anatomia di una Tool Definition
Una tool definition è composta da quattro elementi chiave:
- Nome: segue il pattern
verb_noun(es.search_database,create_file,analyze_code). Il nome deve essere descrittivo è univoco nel contesto dell'agente - Descrizione: la parte più critica della definizione. Il modello usa la descrizione per decidere quando invocare il tool. Deve spiegare cosa fa il tool, quando usarlo e quando NON usarlo
- Parametri: definiti con JSON Schema, specificano tipo, vincoli, valori default e descrizione per ogni parametro accettato dal tool
- Output: il formato atteso della risposta, che aiuta il modello a interpretare i risultati
{
"name": "search_database",
"description": "Search the project database for records matching a query. Use this tool when the user asks about stored data, project records, or needs to find specific entries. Do NOT use this tool for general knowledge questions - those should be answered directly. Supports filtering by date range, status, and category.",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query. Supports full-text search with boolean operators (AND, OR, NOT). Example: 'authentication AND bug NOT resolved'",
"minLength": 1,
"maxLength": 500
},
"table": {
"type": "string",
"description": "The database table to search in",
"enum": ["issues", "pull_requests", "commits", "users", "projects"]
},
"filters": {
"type": "object",
"description": "Optional filters to narrow results",
"properties": {
"status": {
"type": "string",
"enum": ["open", "closed", "in_progress", "resolved"],
"description": "Filter by record status"
},
"date_from": {
"type": "string",
"format": "date",
"description": "Start date for date range filter (ISO 8601)"
},
"date_to": {
"type": "string",
"format": "date",
"description": "End date for date range filter (ISO 8601)"
},
"assignee": {
"type": "string",
"description": "Filter by assigned user"
}
}
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 10,
"minimum": 1,
"maximum": 100
},
"sort_by": {
"type": "string",
"description": "Field to sort results by",
"enum": ["relevance", "date_created", "date_updated", "priority"],
"default": "relevance"
}
},
"required": ["query", "table"]
}
}
Best Practice per le Tool Description
- Sii specifico: "Search the project database" e meglio di "Search for data"
- Includi quando NON usarlo: aiuta il modello a evitare invocazioni inutili
- Fornisci esempi: un esempio concreto di query o parametri vale più di mille parole
- Documenta i limiti: se il tool ha rate limiting, timeout o restrizioni, dillo nella descrizione
- Usa il verb_noun pattern:
search_database,create_issue,delete_file,analyze_code
Input Validation e Sanitization
Quando un modello linguistico genera i parametri per una tool call, non c'è garanzia che i valori siano corretti, sicuri o nel formato atteso. L'input validation è la prima linea di difesa: prima di eseguire qualsiasi operazione, ogni parametro deve essere verificato rispetto al suo schema e sanitizzato per prevenire attacchi injection.
Livelli di Validazione
Un sistema di validazione robusto opera su tre livelli:
- Type Checking: verifica che il tipo di ogni parametro corrisponda allo schema. Una stringa dove è atteso un numero, un array dove è atteso un oggetto, un null dove è atteso un valore non-nullable sono tutti errori da catturare immediatamente
- Bounds Checking: verifica che i valori numerici siano nei range ammessi, che le stringhe rispettino le lunghezze minima e massima, che gli array non superino il numero massimo di elementi
- Constraint Validation: verifica vincoli più complessi come formati (email, URL, date ISO), valori enum, pattern regex, dipendenze tra parametri
from dataclasses import dataclass
from typing import Any
import re
@dataclass
class ValidationError:
field: str
message: str
received_value: Any
class ToolInputValidator:
"""Validatore generico per input di tool call."""
def validate(self, schema: dict, params: dict) -> list[ValidationError]:
errors = []
properties = schema.get("properties", {})
required = schema.get("required", [])
# Verifica campi required
for field in required:
if field not in params or params[field] is None:
errors.append(ValidationError(
field=field,
message=f"Required field '{field}' is missing",
received_value=None
))
# Valida ogni parametro fornito
for field, value in params.items():
if field not in properties:
errors.append(ValidationError(
field=field,
message=f"Unknown field '{field}'",
received_value=value
))
continue
field_schema = properties[field]
errors.extend(self._validate_field(field, value, field_schema))
return errors
def _validate_field(self, field: str, value: Any, schema: dict) -> list[ValidationError]:
errors = []
expected_type = schema.get("type")
# Type checking
type_map = {"string": str, "integer": int, "number": (int, float),
"boolean": bool, "array": list, "object": dict}
if expected_type and not isinstance(value, type_map.get(expected_type, object)):
errors.append(ValidationError(field, f"Expected {expected_type}", value))
return errors # Skip further validation
# String constraints
if expected_type == "string":
if "minLength" in schema and len(value) < schema["minLength"]:
errors.append(ValidationError(field, f"Min length: {schema['minLength']}", value))
if "maxLength" in schema and len(value) > schema["maxLength"]:
errors.append(ValidationError(field, f"Max length: {schema['maxLength']}", value))
if "enum" in schema and value not in schema["enum"]:
errors.append(ValidationError(field, f"Must be one of: {schema['enum']}", value))
if "pattern" in schema and not re.match(schema["pattern"], value):
errors.append(ValidationError(field, f"Must match: {schema['pattern']}", value))
# Numeric constraints
if expected_type in ("integer", "number"):
if "minimum" in schema and value < schema["minimum"]:
errors.append(ValidationError(field, f"Minimum: {schema['minimum']}", value))
if "maximum" in schema and value > schema["maximum"]:
errors.append(ValidationError(field, f"Maximum: {schema['maximum']}", value))
return errors
Protezione da SQL Injection
Quando un tool accetta parametri che verranno usati in query SQL, la protezione da injection è assolutamente critica. Un modello linguistico potrebbe generare parametri contenenti codice SQL malevolo, sia perchè influenzato da contenuto avversariale nel prompt, sia perchè l'utente sta tentando un attacco intenzionale.
La regola fondamentale è semplice e non ammette eccezioni: usare SEMPRE query parameterizzate, mai concatenare stringhe per costruire query SQL. Questo principio vale indipendentemente dal contesto, dalla fiducia nel modello o dalla pressione sulle performance.
# MAI fare questo - vulnerabile a SQL injection
def search_unsafe(query: str, table: str):
sql = f"SELECT * FROM {table} WHERE content LIKE '%{query}%'"
return db.execute(sql) # PERICOLOSO!
# SEMPRE fare questo - query parameterizzata
def search_safe(query: str, table: str):
# Whitelist delle tabelle ammesse
allowed_tables = {"issues", "pull_requests", "commits", "users"}
if table not in allowed_tables:
raise ValueError(f"Table '{table}' not allowed")
sql = "SELECT * FROM " + table + " WHERE content LIKE ?"
return db.execute(sql, (f"%{query}%",)) # Parametro separato
# Ancora meglio: usa un ORM o query builder
def search_orm(query: str, table: str):
model = get_model(table) # Mappa tabella -> modello ORM
return model.objects.filter(content__icontains=query).all()
Attenzione: Injection nei Parametri dei Tool
Non sottovalutare mai il rischio di injection tramite tool parameters. Un attaccante potrebbe inserire istruzioni nel prompt dell'utente che influenzano i parametri generati dal modello. Ad esempio, un prompt come "Cerca nel database; DROP TABLE users; --" potrebbe essere trasformato dal modello in parametri pericolosi se non ci sono validazioni adeguate. La difesa in profondità è essenziale: validazione, sanitizzazione e query parameterizzate a ogni livello.
Output Parsing e Error Recovery
Quando un tool restituisce il suo risultato, l'agente deve essere in grado di interpretarlo correttamente. Un output malformato, un errore imprevisto o un timeout possono interrompere il flusso dell'agente se non gestiti adeguatamente. L'output parsing robusto è fondamentale per la resilienza dell'intero sistema.
Structured Output
Ogni tool dovrebbe restituire output in un formato strutturato e prevedibile. Lo standard prevede un oggetto JSON con campi standardizzati:
@dataclass
class ToolResult:
"""Formato standard per il risultato di un tool."""
success: bool
data: Any = None
error: str | None = None
metadata: dict | None = None
def to_message(self) -> str:
"""Converte il risultato in un messaggio leggibile per il modello."""
if self.success:
if isinstance(self.data, list):
return f"Found {len(self.data)} results:\n" + \
"\n".join(str(item) for item in self.data)
return str(self.data)
else:
return f"Error: {self.error}"
# Esempio di output strutturato
result = ToolResult(
success=True,
data=[
{"id": 1, "title": "Auth bug", "status": "open"},
{"id": 2, "title": "Login issue", "status": "resolved"}
],
metadata={"total_count": 42, "page": 1, "execution_time_ms": 150}
)
Error Recovery Strategies
Quando un tool fallisce, l'agente ha diverse opzioni per recuperare senza interrompere l'interazione con l'utente:
- Retry con backoff esponenziale: per errori transienti (timeout, rate limit, errori di rete), riprova con intervalli crescenti (1s, 2s, 4s, 8s)
- Parametri alternativi: se il tool fallisce con determinati parametri, l'agente può tentare con parametri diversi (query più semplice, range di date più ampio)
- Tool alternativi: se un tool non è disponibile, l'agente può usare un tool alternativo che fornisce informazioni simili
- Degradazione graceful: se nessun tool funziona, l'agente informa l'utente e risponde con le informazioni disponibili nel suo contesto
- Caching dei risultati: i risultati positivi vengono cachati per evitare chiamate ripetute e come fallback in caso di errori successivi
import time
class ToolExecutor:
"""Esecutore di tool con retry e error recovery."""
def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
self.max_retries = max_retries
self.base_delay = base_delay
self.cache: dict = {}
def execute(self, tool_name: str, params: dict) -> ToolResult:
# 1. Controlla la cache
cache_key = f"{tool_name}:{hash(str(sorted(params.items())))}"
if cache_key in self.cache:
cached = self.cache[cache_key]
if time.time() - cached["timestamp"] < 300: # 5 min TTL
return cached["result"]
# 2. Valida gli input
errors = self.validator.validate(tool_name, params)
if errors:
return ToolResult(
success=False,
error=f"Validation failed: {'; '.join(e.message for e in errors)}"
)
# 3. Esegui con retry
last_error = None
for attempt in range(self.max_retries):
try:
result = self._call_tool(tool_name, params)
# 4. Cache il risultato positivo
self.cache[cache_key] = {
"result": result, "timestamp": time.time()
}
return result
except RateLimitError:
delay = self.base_delay * (2 ** attempt)
time.sleep(delay)
last_error = "Rate limit exceeded"
except TimeoutError:
last_error = "Tool execution timed out"
except Exception as e:
last_error = str(e)
break # Non riprovare per errori non transienti
return ToolResult(success=False, error=last_error)
Integrazione API REST
Le API REST sono il punto di integrazione più comune per gli agenti AI. La maggior parte dei servizi web espone API RESTful, e integrare queste API come tool dell'agente permette di accedere a un ecosistema vastissimo di funzionalità: da GitHub per il code management, a Jira per il project tracking, a Slack per la comunicazione, a qualsiasi servizio SaaS con un'API pubblica.
Auto-generazione da OpenAPI Spec
Molte API REST forniscono una OpenAPI specification (ex Swagger) che descrive formalmente tutti gli endpoint, i parametri, i tipi e le risposte. Questa specifica può essere parsata automaticamente per generare le tool definition senza scrivere codice manuale.
import yaml
class OpenAPIToolGenerator:
"""Genera tool definitions da una OpenAPI specification."""
def generate_tools(self, spec_path: str) -> list[dict]:
with open(spec_path) as f:
spec = yaml.safe_load(f)
tools = []
for path, methods in spec.get("paths", {}).items():
for method, details in methods.items():
if method in ("get", "post", "put", "patch", "delete"):
tool = self._endpoint_to_tool(path, method, details, spec)
tools.append(tool)
return tools
def _endpoint_to_tool(self, path, method, details, spec) -> dict:
# Genera nome: GET /users/{id} -> get_user
operation_id = details.get("operationId", f"{method}_{path}")
name = operation_id.replace("-", "_").replace("/", "_")
# Genera parametri da path params, query params e request body
properties = {}
required = []
# Path parameters
for param in details.get("parameters", []):
prop = self._param_to_property(param)
properties[param["name"]] = prop
if param.get("required", False):
required.append(param["name"])
# Request body
if "requestBody" in details:
body_schema = self._resolve_ref(
details["requestBody"]["content"]["application/json"]["schema"],
spec
)
properties["body"] = body_schema
if details["requestBody"].get("required", False):
required.append("body")
return {
"name": name,
"description": details.get("summary", "") + ". " +
details.get("description", ""),
"parameters": {
"type": "object",
"properties": properties,
"required": required
},
"_metadata": {
"http_method": method.upper(),
"path": path,
"base_url": spec.get("servers", [{}])[0].get("url", "")
}
}
Gestione Autenticazione
Integrare API esterne richiede gestire diversi meccanismi di autenticazione. Il framework di tool dell'agente deve supportare i pattern più comuni in modo trasparente, senza esporre credenziali nel contesto del modello linguistico.
Pattern di Autenticazione API
| Metodo | Implementazione | Sicurezza | Caso d'Uso |
|---|---|---|---|
| API Key | Header X-API-Key o query param |
Media | API semplici, servizi interni |
| Bearer Token | Header Authorization: Bearer <token> |
Alta | API RESTful standard, JWT |
| OAuth 2.0 | Flusso di autorizzazione con token refresh | Molto Alta | API di terze parti, accesso delegato |
| mTLS | Certificati client e server bilaterali | Massima | API enterprise, microservizi interni |
Rate Limiting e Retry Logic
Ogni API ha limiti di frequenza che devono essere rispettati. Un agente che chiama API senza gestire il rate limiting rischia di essere bloccato, degradando l'esperienza utente. La gestione deve essere proattiva: tracciare i limiti rimanenti e rallentare prima di raggiungere il limite, non solo reagire dopo.
import time
from collections import defaultdict
class RateLimiter:
"""Rate limiter con token bucket per API esterne."""
def __init__(self):
self.limits: dict[str, dict] = {}
def configure(self, api_name: str, requests_per_minute: int):
self.limits[api_name] = {
"rpm": requests_per_minute,
"tokens": requests_per_minute,
"last_refill": time.time()
}
def acquire(self, api_name: str) -> bool:
"""Tenta di acquisire un token. Restituisce False se rate limited."""
if api_name not in self.limits:
return True
limit = self.limits[api_name]
now = time.time()
elapsed = now - limit["last_refill"]
# Refill tokens proporzionalmente al tempo trascorso
refill = elapsed * (limit["rpm"] / 60.0)
limit["tokens"] = min(limit["rpm"], limit["tokens"] + refill)
limit["last_refill"] = now
if limit["tokens"] >= 1:
limit["tokens"] -= 1
return True
return False
def wait_time(self, api_name: str) -> float:
"""Tempo di attesa stimato prima del prossimo token disponibile."""
if api_name not in self.limits:
return 0
limit = self.limits[api_name]
if limit["tokens"] >= 1:
return 0
return (1 - limit["tokens"]) * (60.0 / limit["rpm"])
Integrazione GraphQL
GraphQL offre un'alternativa flessibile alle API REST per l'integrazione con gli agenti AI. A differenza di REST, dove ogni endpoint restituisce una struttura fissa, GraphQL permette al client di specificare esattamente quali campi desidera nella risposta. Questo è particolarmente vantaggioso per gli agenti perchè riduce la quantità di dati trasferiti e i token necessari per processare la risposta.
Query e Mutation come Tools
In GraphQL, le operazioni si dividono in query (lettura) e mutation (scrittura). Ogni query o mutation può essere esposta come un tool separato dell'agente, con i parametri GraphQL mappati ai parametri del tool.
class GraphQLToolAdapter:
"""Adatta operazioni GraphQL a tool per l'agente."""
def __init__(self, endpoint: str, headers: dict = None):
self.endpoint = endpoint
self.headers = headers or {}
def create_query_tool(self, name: str, query: str, variables_schema: dict) -> dict:
"""Crea un tool da una query GraphQL."""
return {
"name": name,
"description": f"Execute GraphQL query: {name}",
"parameters": {
"type": "object",
"properties": variables_schema
},
"_executor": lambda params: self._execute(query, params)
}
def _execute(self, query: str, variables: dict) -> ToolResult:
import requests
response = requests.post(
self.endpoint,
json={"query": query, "variables": variables},
headers=self.headers
)
data = response.json()
if "errors" in data:
return ToolResult(
success=False,
error=str(data["errors"])
)
return ToolResult(success=True, data=data.get("data"))
# Esempio di utilizzo
adapter = GraphQLToolAdapter("https://api.example.com/graphql")
tool = adapter.create_query_tool(
name="get_project_issues",
query="""
query GetIssues($projectId: ID!, $status: String) {
project(id: $projectId) {
name
issues(status: $status) {
id
title
assignee { name }
priority
}
}
}
""",
variables_schema={
"projectId": {"type": "string", "description": "Project ID"},
"status": {"type": "string", "enum": ["OPEN", "CLOSED", "IN_PROGRESS"]}
}
)
Schema Introspection per Tool Generation
Una delle caratteristiche più potenti di GraphQL è la schema introspection: la capacità di interrogare lo schema stesso per scoprire tutti i tipi, le query e le mutation disponibili. Questo permette di generare automaticamente le tool definition dall'API GraphQL senza documentazione esterna, analizzando lo schema a runtime.
Database Tools
I database tools permettono all'agente di interrogare e modificare database relazionali e NoSQL. Sono tra i tool più potenti e al contempo più rischiosi: una query errata può esporre dati sensibili, corrompere record o causare problemi di performance.
Architettura di Sicurezza per Database Tools
L'accesso al database deve essere mediato da un sistema di sicurezza a più livelli:
- Read-only vs Read-write permissions: per default, i tool di database dovrebbero avere accesso in sola lettura. Le operazioni di scrittura devono richiedere autorizzazione esplicita e conferma da parte dell'utente
- Query allowlist: limita i tipi di query eseguibili. Solo SELECT, niente DROP, ALTER, o TRUNCATE
- Row limit: imponi sempre un LIMIT alle query per evitare di restituire milioni di righe e saturare il contesto
- Table access control: definisci una whitelist di tabelle accessibili, escludendo tabelle sensibili come quelle con password o dati finanziari
- Audit logging: registra ogni query eseguita, con timestamp, utente e risultato, per accountability e debug
class SecureDatabaseTool:
"""Tool di database con sicurezza a più livelli."""
ALLOWED_TABLES = {"issues", "projects", "sprints", "tasks", "comments"}
MAX_ROWS = 100
BLOCKED_KEYWORDS = {"DROP", "ALTER", "TRUNCATE", "DELETE", "INSERT", "UPDATE"}
def __init__(self, db_connection, read_only: bool = True):
self.db = db_connection
self.read_only = read_only
self.audit_log = []
def execute_query(self, query: str, params: tuple = ()) -> ToolResult:
# 1. Sanitization: controlla per keyword pericolose
query_upper = query.upper().strip()
if self.read_only:
if not query_upper.startswith("SELECT"):
return ToolResult(False, error="Only SELECT queries allowed in read-only mode")
for keyword in self.BLOCKED_KEYWORDS:
if keyword in query_upper:
return ToolResult(False, error=f"Blocked keyword: {keyword}")
# 2. Verifica le tabelle accedute
tables_in_query = self._extract_tables(query)
unauthorized = tables_in_query - self.ALLOWED_TABLES
if unauthorized:
return ToolResult(False, error=f"Access denied to tables: {unauthorized}")
# 3. Aggiungi LIMIT se mancante
if "LIMIT" not in query_upper:
query = query.rstrip(";") + f" LIMIT {self.MAX_ROWS}"
# 4. Esegui con parametri (MAI concatenazione di stringhe)
try:
cursor = self.db.execute(query, params)
columns = [desc[0] for desc in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
# 5. Audit log
self.audit_log.append({
"query": query, "params": params,
"rows_returned": len(rows),
"timestamp": time.time()
})
return ToolResult(
success=True,
data=rows,
metadata={"columns": columns, "row_count": len(rows)}
)
except Exception as e:
return ToolResult(False, error=f"Query error: {str(e)}")
Custom Tool Framework
In un sistema di produzione con decine o centinaia di tool, serve un framework che standardizzi la creazione, la registrazione e l'esecuzione dei tool. Un buon framework riduce il boilerplate, garantisce consistenza e facilità il testing.
Pattern per un Framework Riutilizzabile
Il framework si basa su un decoratore @tool che trasforma una funzione Python
ordinaria in un tool registrabile dall'agente. Il decoratore si occupa automaticamente di
generare la tool definition dal docstring e dai type hints, validare gli input, gestire gli
errori e formattare l'output.
import inspect
from functools import wraps
from typing import get_type_hints
class ToolRegistry:
"""Registry centralizzato per tutti i tool dell'agente."""
def __init__(self):
self._tools: dict[str, dict] = {}
def tool(self, name: str = None, description: str = None):
"""Decoratore per registrare una funzione come tool."""
def decorator(func):
tool_name = name or func.__name__
tool_desc = description or func.__doc__ or "No description"
# Genera lo schema dei parametri dai type hints
hints = get_type_hints(func)
sig = inspect.signature(func)
properties = {}
required = []
for param_name, param in sig.parameters.items():
if param_name == "self":
continue
param_type = hints.get(param_name, str)
prop = self._type_to_schema(param_type)
# Usa il docstring per la descrizione del parametro
prop["description"] = f"Parameter: {param_name}"
properties[param_name] = prop
if param.default is inspect.Parameter.empty:
required.append(param_name)
# Registra il tool
self._tools[tool_name] = {
"definition": {
"name": tool_name,
"description": tool_desc,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
},
"executor": func
}
@wraps(func)
def wrapper(*args, **kwargs):
try:
result = func(*args, **kwargs)
return ToolResult(success=True, data=result)
except Exception as e:
return ToolResult(success=False, error=str(e))
return wrapper
return decorator
def get_definitions(self) -> list[dict]:
"""Restituisce le definizioni di tutti i tool registrati."""
return [t["definition"] for t in self._tools.values()]
def execute(self, tool_name: str, params: dict) -> ToolResult:
"""Esegue un tool per nome con i parametri forniti."""
if tool_name not in self._tools:
return ToolResult(False, error=f"Unknown tool: {tool_name}")
executor = self._tools[tool_name]["executor"]
return executor(**params)
# Utilizzo del framework
registry = ToolRegistry()
@registry.tool(
name="analyze_code",
description="Analyze source code for potential issues, complexity, and style."
)
def analyze_code(code: str, language: str, checks: list[str] = None) -> dict:
"""Analizza il codice sorgente cercando problemi e suggerimenti."""
results = {
"issues": [],
"complexity_score": 0,
"suggestions": []
}
# ... logica di analisi ...
return results
@registry.tool(
name="search_documentation",
description="Search the project documentation for relevant articles and guides."
)
def search_documentation(query: str, section: str = None, limit: int = 5) -> list:
"""Cerca nella documentazione del progetto."""
# ... logica di ricerca ...
return results
Tool Discovery e Dynamic Registration
Negli agenti avanzati, i tool disponibili non sono fissi: possono cambiare a runtime in base al contesto, alle autorizzazioni dell'utente o alla disponibilità dei servizi esterni. Il tool discovery è il meccanismo che permette all'agente di scoprire nuovi tool e registrarli dinamicamente senza riavviare il sistema.
Capability Inference
Quando l'agente ha accesso a molti tool (decine o centinaia), non è pratico includerli tutti nel prompt: occuperebbero troppi token e confonderebbero il modello. La capability inference seleziona dinamicamente solo i tool rilevanti per la query corrente, basandosi sulla similarità semantica tra la descrizione del tool e la richiesta dell'utente.
class DynamicToolSelector:
"""Seleziona i tool più rilevanti per la query corrente."""
def __init__(self, registry: ToolRegistry, embedding_model):
self.registry = registry
self.embedding_model = embedding_model
self._tool_embeddings: dict[str, list[float]] = {}
self._index_tools()
def _index_tools(self):
"""Indicizza le descrizioni dei tool per la ricerca semantica."""
for tool in self.registry.get_definitions():
text = f"{tool['name']}: {tool['description']}"
self._tool_embeddings[tool["name"]] = \
self.embedding_model.embed(text)
def select_tools(self, query: str, max_tools: int = 10) -> list[dict]:
"""Seleziona i tool più rilevanti per la query."""
query_embedding = self.embedding_model.embed(query)
# Calcola la similarità con ogni tool
scores = []
for tool_name, tool_embedding in self._tool_embeddings.items():
similarity = cosine_similarity(query_embedding, tool_embedding)
scores.append((tool_name, similarity))
# Ordina per similarità decrescente
scores.sort(key=lambda x: x[1], reverse=True)
# Restituisci i top-k tool
selected_names = [name for name, _ in scores[:max_tools]]
return [
t for t in self.registry.get_definitions()
if t["name"] in selected_names
]
Vantaggi del Tool Discovery Dinamico
- Scalabilità: supporta centinaia di tool senza saturare la context window
- Precisione: il modello vede solo i tool rilevanti, riducendo la confusione e gli errori
- Estensibilita: nuovi tool possono essere aggiunti a runtime senza modificare il codice dell'agente
- Sicurezza: i tool possono essere filtrati in base alle autorizzazioni dell'utente
- Riduzione dei costi: meno tool nel prompt significa meno token consumati per ogni chiamata
Streaming e Long-Running Tools
Non tutti i tool completano la loro esecuzione in millisecondi. Alcune operazioni richiedono tempo significativo: analisi di grandi codebase, deploy di applicazioni, generazione di report complessi, esecuzione di test suite. Per questi casi, il framework deve supportare esecuzione asincrona con progress reporting.
Pattern per Tool Long-Running
import asyncio
from enum import Enum
class ToolStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class AsyncToolExecutor:
"""Esecutore asincrono per tool long-running."""
def __init__(self, timeout_seconds: int = 300):
self.timeout = timeout_seconds
self.tasks: dict[str, dict] = {}
async def execute_async(self, tool_name: str, params: dict,
on_progress=None) -> str:
"""Avvia l'esecuzione asincrona e restituisce un task ID."""
task_id = f"task_{tool_name}_{int(time.time())}"
self.tasks[task_id] = {
"status": ToolStatus.PENDING,
"progress": 0,
"result": None,
"started_at": time.time()
}
# Avvia il task in background
asyncio.create_task(
self._run_with_timeout(task_id, tool_name, params, on_progress)
)
return task_id
async def _run_with_timeout(self, task_id, tool_name, params, on_progress):
self.tasks[task_id]["status"] = ToolStatus.RUNNING
try:
result = await asyncio.wait_for(
self._execute(tool_name, params, task_id, on_progress),
timeout=self.timeout
)
self.tasks[task_id]["status"] = ToolStatus.COMPLETED
self.tasks[task_id]["result"] = result
except asyncio.TimeoutError:
self.tasks[task_id]["status"] = ToolStatus.FAILED
self.tasks[task_id]["result"] = ToolResult(
False, error=f"Timeout after {self.timeout}s"
)
except Exception as e:
self.tasks[task_id]["status"] = ToolStatus.FAILED
self.tasks[task_id]["result"] = ToolResult(False, error=str(e))
def get_status(self, task_id: str) -> dict:
"""Ottieni lo stato corrente di un task."""
if task_id not in self.tasks:
return {"error": "Task not found"}
task = self.tasks[task_id]
return {
"status": task["status"].value,
"progress": task["progress"],
"elapsed": time.time() - task["started_at"],
"result": task["result"] if task["status"] == ToolStatus.COMPLETED else None
}
async def cancel(self, task_id: str) -> bool:
"""Annulla un task in esecuzione."""
if task_id in self.tasks:
self.tasks[task_id]["status"] = ToolStatus.CANCELLED
return True
return False
Progress Callbacks
Per i tool che richiedono tempo, i progress callback sono essenziali per mantenere l'utente informato. L'agente può mostrare lo stato di avanzamento in tempo reale, permettendo all'utente di decidere se attendere o annullare l'operazione.
- Percentuale di completamento: per operazioni con progresso quantificabile (analisi di file, batch processing)
- Messaggi di stato: per operazioni a fasi discrete ("Connessione al server...", "Analisi dei risultati...", "Generazione del report...")
- Risultati parziali: per operazioni che producono output incrementale (streaming di risultati di ricerca, analisi progressiva)
Pattern di Composizione dei Tool
La potenza degli agenti non risiede nei singoli tool, ma nella capacità di comporli in flussi di lavoro complessi. Un agente maturo può concatenare più tool per completare task che nessun tool singolo potrebbe gestire.
Esempio di Composizione: Analisi e Fix di un Bug
Quando un utente segnala un bug, l'agente può orchestrare automaticamente una sequenza di tool call:
search_database: cerca nel bug tracker se il bug è già stato segnalatosearch_codebase: trova i file di codice rilevanti per il componente affettoanalyze_code: analizza il codice trovato per identificare la causa del buggenerate_fix: genera una patch suggerita basata sull'analisirun_tests: esegue i test per verificare che la fix non introduca regressionicreate_pull_request: crea una PR con la fix e una descrizione dettagliata
Ogni tool utilizza l'output del tool precedente come input, creando una pipeline automatizzata che trasforma una segnalazione di bug in una pull request verificata.
Conclusioni
Il tool calling è il meccanismo che trasforma gli agenti AI da generatori di testo passivi a orchestratori attivi capaci di agire nel mondo reale. La qualità del tool calling dipende da tre pilastri: definizioni precise con JSON Schema dettagliato, validazione rigorosa degli input per prevenire errori e attacchi, e gestione robusta degli errori con retry e fallback appropriati.
Abbiamo visto come integrare API REST, GraphQL e database in modo sicuro, come costruire un framework di tool riutilizzabili con discovery dinamico, e come gestire operazioni long-running con esecuzione asincrona e progress reporting. Questi pattern costituiscono la base su cui costruire agenti di produzione affidabili e scalabili.
Nel prossimo articolo affronteremo il testing degli agenti AI: come testare i flussi di tool calling, come simulare risposte del modello, come misurare la qualità delle decisioni dell'agente e come implementare test di regressione per garantire che le modifiche non introducano comportamenti inattesi.