Custom Tools

En esta página se detallan claramente las funciones disponibles en Saptiva Tools, explicando su propósito, entradas necesarias y salidas generadas.

---

1. Búsqueda en Wikipedia

wikipedia_search(query: str)

Realiza búsquedas directamente en Wikipedia sin necesidad de herramientas base adicionales.

Parámetros:

• query: Texto que deseas buscar en Wikipedia.

Requisitos:

pip install wikipedia

Retorna:

• Contenido relevante obtenido desde Wikipedia.

Ejemplo:

import asyncio

from saptiva_agents.tools import wikipedia_search

from saptiva_agents.ui import Console
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.base import SaptivaAIChatCompletionClient

model_client = SaptivaAIChatCompletionClient(model="Saptiva Legacy", api_key="va-ai-...")
tool = wikipedia_search
agent = AssistantAgent(
    name="wikipedia_agent",
    model_client=model_client,
    tools=[tool],
    system_message="Eres un asistente servicial. Puedes llamar 'tools' para ayudar al usuario.",
    reflect_on_tool_use=True,
    model_client_stream=True,
)

async def main() -> None:
    await Console(
        agent.run_stream(task="Donde se celebró el Campeonato Europeo de Lucha de 2025"),
        output_stats=False,
    )

    await model_client.close()

if __name__ == "__main__":
    asyncio.run(main())

Salida:

---------- TextMessage (user) ----------
Donde se celebró el Campeonato Europeo de Lucha de 2025
---------- ToolCallRequestEvent (wikipedia_agent) ----------
[FunctionCall(id='call_u1l1d53e', arguments='{"query":"Campeonato Europeo de Lucha de 2025"}', name='wikipedia_search')]
---------- ToolCallExecutionEvent (wikipedia_agent) ----------
[FunctionExecutionResult(content='Page: Campeonato Europeo de Lucha de 2025\nSummary: El LXXVI Campeonato Europeo de Lucha se celebró en Bratislava (Eslovaquia) entre el 7 y el 13 de abril de 2025 bajo la organización de United World Wrestling (UWW) y la Federación Eslovaca de Lucha.[1]\u200b\nLas competiciones se realizaron en el recinto X-Bionic Sphere de la capital eslovaca.\n\nPage: Campeonato Europeo de Lucha\nSummary: El Campeonato Europeo de Lucha es la competición del deporte de lucha más importante a nivel europeo. Es organizado desde 1911 por la Federación Internacional de Luchas Asociadas (FILA). Actualmente se realiza cada a', name='wikipedia_search', call_id='call_u1l1d53e', is_error=False)]
---------- ModelClientStreamingChunkEvent (wikipedia_agent) ----------
El Campeonato Europeo de Lucha de 2025 se celebró en Bratislava, Eslovaquia. Las competiciones tuvieron lugar en el recinto X-Bionic Sphere de la capital eslovaca entre el 7 y el 13 de abril de 2025.

Process finished with exit code 0

---

2. Consultas al Bot de Saptiva

saptiva_bot_query(userMessage: str)

Envía una consulta al bot de Saptiva y obtiene una respuesta.

Parámetros:

• userMessage: Mensaje o consulta que deseas enviar al bot.

Retorna:

• Respuesta del bot en formato JSON.

Ejemplo:

import asyncio
import json

from saptiva_agents import SAPTIVA_LEGACY

from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents.tools import (
    # Para este ejemplo usaremos la saptiva_bot_query
    saptiva_bot_query,
)

model_client = SaptivaAIChatCompletionClient(
    model=SAPTIVA_LEGACY,
    api_key="TU_SAPTIVA_API_KEY"
)

agent = AssistantAgent(
    "sample_agent",
    model_client=model_client,
    system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
    tools=[saptiva_bot_query]
)

async def run():
    result = await agent.run(task="invoka `saptiva_bot_query` ¿Cuáles son los productos de Saptibank?")
    result_json = json.dumps(result.messages[2].content[0].content, ensure_ascii=False, indent=2)    
    print("Result:")
    print(result_json)

# Corre el agente
if __name__ == "__main__":
    asyncio.run(run())

Salida:

Result:
"{'response': 'Saptibank ofrece una variedad de productos financieros para satisfacer las necesidades diversas de nuestros clientes. Aquí te menciono algunos de ellos:\\n\\n1. Cuentas de Ahorro: Estas cuentas ofrecen una forma segura de ahorrar dinero y ganar intereses.\\n\\n2. Cuentas Corrientes: Son útiles para manejar transacciones diarias como depósitos, retiros, y pagos.\\n\\n3. Tarjetas de Crédito: Ofrecemos varias tarjetas de crédito con diferentes beneficios y características, que se adaptan a las necesidades de nuestros clientes.\\n\\n4. Préstamos: Ofrecemos préstamos personales, préstamos para autos, y préstamos hipotecarios para ayudar a nuestros clientes a financiar sus necesidades.\\n\\n5. Inversiones: Tenemos opciones de inversión que pueden ayudar a nuestros clientes a crecer su dinero a largo plazo.\\n\\n6. Seguros: Ofrecemos una variedad de seguros para ayudar a proteger a nuestros clientes y sus bienes.\\n\\nPor favor, dime si necesitas información más detallada sobre alguno de estos productos.'}"

Process finished with exit code 0

---

3. Manejo de CSV

upload_csv(uniqueID: str, file: str)

Sube un archivo CSV codificado en Base64 al servicio analizador Saptibank (Saptibank analyzer service).

Parámetros:

• uniqueID: Identificador único para la operación.

• file: Archivo CSV codificado en Base64.

Retorna:

• Respuesta JSON del analizador Saptibank.

Ejemplo:

import asyncio
import base64
import os

from saptiva_agents.tools import upload_csv
from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents import SAPTIVA_LEGACY

async def run():
    file_path = os.path.join(os.path.dirname(__file__), "employees.csv")
    with open(file_path, "rb") as file:
        csv_content = file.read()

    base64_encoded = base64.b64encode(csv_content).decode("utf-8") 

    model_client = SaptivaAIChatCompletionClient(model=SAPTIVA_LEGACY, api_key="TU_SAPTIVA_API_KEY")

    agent = AssistantAgent(
        "csv_agent",
        model_client=model_client,
        system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
        tools=[upload_csv]
    )

    result = await agent.run(task=f"llama a `upload_csv` con estos parametros: `uniqueID`: '123456789', `file`: {base64_encoded}")
    print(result)

if __name__ == "__main__":
    asyncio.run(run())

---

4. Consulta CFDI/CURP

consultar_cfdi(url: str)

Consume un CFDI usando la API de Vulcanics, con lógica de reintento automático.

Parámetros:

• url: URL del CFDI a consultar.

Retorna:

• Información obtenida del CFDI.

Ejemplo:

import asyncio

from saptiva_agents.tools import consultar_cfdi
from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents import SAPTIVA_LEGACY

async def run():
    model_client = SaptivaAIChatCompletionClient(model=SAPTIVA_LEGACY, api_key="TU_SAPTIVA_API_KEY")

    agent = AssistantAgent(
        "cfdi_agent",
        model_client=model_client,
        system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
        tools=[consultar_cfdi]
    )

    result = await agent.run(task="llama a `consultar_cfdi` con este parametro: https://verificacfdi.facturaelectronica.sat.gob.mx/?id=27381870-40DD-4EED-B753-9A3901595268&re=MAHM970519LI2&rr=FFA000411RIA&tt=21762.93&fe=KblKLQ==")
    print(result)

if __name__ == "__main__":
    asyncio.run(run())

---

consultar_curp_post(curp: str)

Envía una CURP mediante POST y obtiene un identificador de consulta. Permite consumir fase de envío de CURPS mediante la plataforma Vulcanics.

Parámetros:

• curp: CURP válida.

Retorna:

• Identificador (ID) generado por la API para realizar consultas posteriores.

Ejemplo:

import asyncio

from saptiva_agents.tools import consultar_curp_post
from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents import SAPTIVA_LEGACY

async def run():
    model_client = SaptivaAIChatCompletionClient(model=SAPTIVA_LEGACY, api_key="TU_SAPTIVA_API_KEY")

    agent = AssistantAgent(
        "curp_post_agent",
        model_client=model_client,
        system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
        tools=[consultar_curp_post]
    )

    result = await agent.run(task="llama a `consultar_curp_post` con este parametro: 'curp: LOGG900311HDFZML05'")
    print(result)

if __name__ == "__main__":
    asyncio.run(run())

---

consultar_curp_get(curp_id: str)

Obtiene los detalles asociados a una CURP mediante un identificador generado previamente. Permite consumir fase de extracción de CURPS mediante la plataforma Vulcanics.

Parámetros:

• curp_id: ID obtenido previamente.

Retorna:

• Información detallada de la CURP consultada.

Ejemplo:

import asyncio

from saptiva_agents.tools import consultar_curp_get
from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents import SAPTIVA_LEGACY

async def run():
    model_client = SaptivaAIChatCompletionClient(model=SAPTIVA_LEGACY, api_key="TU_SAPTIVA_API_KEY")

    agent = AssistantAgent(
        "curp_get_agent",
        model_client=model_client,
        system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
        tools=[consultar_curp_get]
    )

    result = await agent.run(task="llama a `consultar_curp_get` con este parametro: LOGG900311HDFZML05")
    print(result)

if __name__ == "__main__":
    asyncio.run(run())

---

get_verify_sat(url: str)

Realiza una verificación SAT mediante la plataforma Vulcanics.

Parámetros:

• url: URL del SAT a verificar.

Retorna:

• Resultado de la verificación.

Ejemplo:

import asyncio

from saptiva_agents.tools import get_verify_sat
from saptiva_agents.base import SaptivaAIChatCompletionClient
from saptiva_agents.agents import AssistantAgent
from saptiva_agents import SAPTIVA_LEGACY

async def run():
    model_client = SaptivaAIChatCompletionClient(model=SAPTIVA_LEGACY, api_key="TU_SAPTIVA_API_KEY")

    agent = AssistantAgent(
        "sat_verify_agent",
        model_client=model_client,
        system_message="You are an agent that helps users by the usage of the `tools` you have been provided with.",
        tools=[get_verify_sat]
    )

    result = await agent.run(task="llama a `get_verify_sat` con este parametro: https://www.sat.gob.mx/")
    print(result)

if __name__ == "__main__":
    asyncio.run(run())

---

5. Extractor PDF

obtener_texto_en_documento(doc_type: str, document: str, key: str="")

Extrae texto de documentos PDF usando el servicio Saptiva API Extractor.

Parámetros:

• doc_type: Tipo de documento (por ejemplo: pdf).

• document: Documento codificado en Base64.

• key (Opcional): Llave API del extractor. Si no se proporciona, se toma del entorno (SAPTIVA_API_KEY).

Retorna:

• JSON con texto extraído del documento.

Ejemplo:

import asyncio
import base64
import os
from saptiva_agents.tools import obtener_texto_en_documento


async def run():
    file_path = os.path.join(os.path.dirname(__file__), "hello.pdf")
    with open(file_path, "rb") as file:
        pdf_content = file.read()

    base64_encoded = base64.b64encode(pdf_content).decode("utf-8") 
    try:
        result = await obtener_texto_en_documento(
            doc_type="pdf",
            document=base64_encoded,
            key="TU_SAPTIVA_API_KEY"
        )
        print(f"✅ Success: {result}")
    except Exception as e:
        print(f"❌ Error: {e}")

if __name__ == "__main__":
    asyncio.run(run())