Files

T

2026-09-24 05:19:06 -04:00

9.8 KiB

Raw Permalink Blame History

ApiPilot

Libreria Python generica che trasforma endpoint di un'app FastAPI in tool eseguibili da agenti AI, senza riferimenti hardcoded a domini o progetti specifici.

Che cosa fa ApiPilot

Scansiona l'OpenAPI della tua app FastAPI e registra solo gli endpoint marcati con x-apipilot.enabled: true
Genera definizioni di tool nel formato function calling compatibile OpenAI (nome, descrizione, schema JSON parametri per path / query / body)
Esegue i tool validando gli argomenti e applicando l'autenticazione e i permessi della tua app ospite
Gestisce timeout, risposte troppo grandi, errori strutturati e risultati vuoti
Permette di registrare tool composti Python (multi-step) non legati a un singolo endpoint
Separa il nucleo dal codice dei provider LLM, con adapter iniziale per il formato OpenAI (incluso tool_call_id)
Opzionalmente espone catalogo ed esecuzione via HTTP con autenticazione della tua app

Installazione

# Crea e attiva l'ambiente virtuale (consigliato)
python -m venv .venv
source .venv/bin/activate           # Linux/macOS
# .\.venv\Scripts\activate         # Windows PowerShell

# Installa ApiPilot (modalità editabile + dipendenze test/dev
pip install -e ".[test,dev]"

Senza venv (sconsigliato, a seconda della policy PEP 668):

pip install -e ".[test]" --break-system-packages

Requisiti minimi: Python 3.10+, FastAPI 0.100+, Pydantic v2.

Utilizzo base

1. Abilita un endpoint con `x-apipilot`

Aggiungi openapi_extra={"x-apipilot": {...}} ai decorator delle rotte che vuoi esporre come tool:

from fastapi import APIRouter

router = APIRouter()

@router.get(
    "/items",
    summary="Elenca gli elementi",
    openapi_extra={
        "x-apipilot": {
            "enabled": True,
            "name": "list_items",
            "description": (
                "Cerca gli elementi disponibili. Usalo quando "
                "l'utente chiede un elenco o cerca elementi "
                "di una categoria."
            ),
        }
    },
)
async def list_items(category: str | None = None, limit: int = 10):
    ...

2. Installa ApiPilot sulla tua app

from fastapi import FastAPI
from apipilot.fastapi import install_apipilot

app = FastAPI()

def get_current_user(authorization: str | None = None):
    # La tua logica di autenticazione esistente
    ...

pilot = install_apipilot(
    app,
    get_current_user=get_current_user,
)

3. Ottieni i tool disponibili per l'utente

tools = await pilot.tools_for(user)
# Ogni tool è già in formato:
# {"name": "...", "description": "...", "parameters": {...}}

4. Esegui un tool

result = await pilot.execute(
    user=user,
    name="list_items",
    arguments={"category": "active"},
)
# result: {"success": bool, "data": Any, "error": dict|None, "meta": dict}

5. Integra con un agente AI (es. OpenAI function calling)

from apipilot.adapters.openai import OpenAIAdapter

# conversione -> messaggio tools per OpenAI
openai_tools = OpenAIAdapter.to_tools(tools)

# ricevi la tool_call dal modello e parsala
tool_call_id, name, args = OpenAIAdapter.parse_tool_call(assistant_msg.tool_calls[0])

# esegui
resp = await pilot.execute(user=user, name=name, arguments=args)

# costruisci il messaggio di risposta per il ciclo successivo
tool_result = OpenAIAdapter.make_tool_result(tool_call_id, name, resp)

Formato di `x-apipilot`

Ogni operazione abilitata dichiara in openapi_extra["x-apipilot"]:

Campo	Tipo	Obbligatorio	Default	Note
`enabled`	`bool`	Sì	`false`	Deve essere `true` per essere preso in carico.
`name`	`string`	Consigliato	da `operationId` o route	Nome univoco del tool, usato anche da `execute`.
`description`	`string`	No	`summary` endpoint	Descrizione per l'agente AI (influisce sulla scelta).
`timeout`	`number`	No	`default_timeout` (30s)	Secondi oltre i quali la chiamata viene interrotta.
`max_response_size`	`number`	No	`1_000_000` (1 MB)	Limite byte risultato JSON oltre il quale viene troncato.

Nomi duplicati, configurazioni incomplete o schemi non supportati ($ref non risolti fuori standard) sollevano errori chiari in fase di installazione.

Autorizzazione (controllo per-tool)

def allow_tool(user, record) -> bool:
    if user and user.get("role") != "admin" and record.name == "create_item":
        return False
    return True

pilot = install_apipilot(app, get_current_user=get_current_user, allow_tool=allow_tool)

I tool non ammessi non compaiono in tools_for e execute solleva ToolNotFoundError se richiamati.

Tool composti (multi-step)

from apipilot.tool import CompositeTool

async def find_cheapest(args, user):
    res = await pilot.execute(user=user, name="list_items", arguments=args)
    if not res["success"]:
        return {"error": res["error"]}
    items = (res["data"] or {}).get("items", [])
    if not items:
        return None
    return {"cheapest": min(items, key=lambda it: it["price"])}

pilot.add_tool(
    CompositeTool(
        name="find_cheapest_item",
        description="Trova l'elemento più economico nella categoria.",
        parameters_schema={
            "type": "object",
            "properties": {"category": {"type": "string"}},
            "required": [],
        },
        run_fn=find_cheapest,
    )
)

Endpoints HTTP opzionali

Passa expose_http_endpoints=True:

pilot = install_apipilot(app, get_current_user=get_current_user, expose_http_endpoints=True, http_prefix="/apipilot")

GET /apipilot/tools — catalogo tool per l'utente autenticato.
POST /apipilot/execute payload {"name": "...", "arguments": {...}}.

Se require_auth_for_http=True (default) e get_current_user è fornito, gli endpoint HTTP usano la tua funzione di autenticazione.

Struttura risposta `ExecuteResponse`

{
    "success": True | False,
    "data": Any | None,
    "error": {
        "code": "INVALID_ARGUMENTS" | "EXECUTION_ERROR" | "TIMEOUT" | "UNEXPECTED_EXECUTION_ERROR",
        "message": str,
        # altri campi specifici (status_code, endpoint, details, ...)
    } | None,
    "meta": {
        "tool": str,
        "truncated"?: True,
        "max_response_size"?: int,
    }
}

Sicurezza — garanzie di ApiPilot

Il modello non sceglie URL, metodo HTTP o credenziali: ogni tool mappa in modo statico una rotta registrata.
Header tecnici e credenziali non sono esposti nello schema dei parametri del tool (i parametri in: header sono saltati).
Autorizzazione a due livelli: controllo allow_tool(user, record) prima ancora della validazione argomenti, più le dipendenze native FastAPI sull'endpoint target.
Validazione schema argomenti tramite jsonschema prima di invocare qualsiasi funzione endpoint.
Timeout per esecuzione e capped size delle risposte per evitare injection di contesto eccessivo.
Nomi tool univoci: duplicati intercettati in fase di installazione.

Eseguire i test

pip install -e ".[test]"
pytest -q

Coverage di riferimento: selezione operazioni, generazione parametri, autorizzazione, rifiuto tool sconosciuti, argomenti invalidi, gestione errori, timeout, troncamento risposte grandi, tool composti, endpoints HTTP.

Demo backend incluso

Servizio completo (start/stop/restart/status/logs):

./demo.sh start     # avvia in background, crea .venv se serve
./demo.sh status    # verifica stato (PID e log path)
./demo.sh logs 50   # ultime 50 righe di log
./demo.sh restart   # stop + start
./demo.sh stop      # arresto graceful (con fallback SIGKILL)
./demo.sh help      # elenco comandi

# URL  : http://127.0.0.1:9292/docs
# Auth : Demo senza autenticazione obbligatoria.
#        Senza header: default user=demo-user, ruolo=admin
#        Con header Authorization: Bearer admin-token / Bearer user-token per testare altri utenti
# Files: .demo.pid (PID), .demo.log (log), .venv/ (ambiente)

Modo veloce, primo piano (con venv auto):

./run_demo.sh
# → crea .venv se manca, installa dipendenze, avvia su http://127.0.0.1:9292

Modo manuale con venv:

# Crea venv e installa (solo se non già fatto)
python -m venv .venv
source .venv/bin/activate
pip install -e ".[test]"

# Avvia il demo
python -m examples.demo_backend
# → http://127.0.0.1:9292/docs
# Demo senza autenticazione obbligatoria.
# Senza header Authorization usa default user=demo-user (ruolo admin)
# Con Authorization: Bearer admin-token / Bearer user-token puoi testare altri utenti

Vedi examples/demo_backend.py per uno scenario completo (catalogo prodotti, ordini, tool composto di raccomandazione, demo ciclo agente).

Moduli

apipilot/core.py — classe ApiPilot, autorizzazione, esecuzione, gestione errori/size/timeout.
apipilot/fastapi.py — install_apipilot, endpoints HTTP opzionali.
apipilot/openapi_parser.py — parsing OpenAPI, generazione schema JSON, deduplica/valida.
apipilot/validation.py — validazione argomenti via jsonschema.
apipilot/tool.py — CompositeTool e protocollo PilotTool.
apipilot/errors.py — gerarchia eccezioni.
apipilot/types.py — TypedDict e dataclass condivisi.
apipilot/adapters/openai.py — adapter per function calling OpenAI (to_tools, parse_tool_call, make_tool_result con tool_call_id).

Modifiche & versionamento

Vedere CHANGELOG.md per il tracciamento per data delle modifiche.

9.8 KiB Raw Permalink Blame History