Herramientas custom del Open Agents SDK via Arcade.dev

La Open Agent Platform de LangChain se conecta a herramientas mediante el Model Context Protocol (MCP). Arcade.dev ofrece la infraestructura para construir, desplegar y servir herramientas personalizadas como servidores MCP que los agentes de OAP pueden invocar.

Esta guía cubre la construcción de herramientas personalizadas con el SDK de Arcade, su despliegue como servidores MCP y su integración en la Open Agent Platform de LangChain.

Resumen de arquitectura

El flujo de integración funciona así:

Agente LangGraph → Protocolo MCP → Servidor MCP de Arcade → Ejecución de herramienta personalizada → Respuesta

Componentes:

Agentes LangGraph: Corren en LangGraph Platform, gestionan la conversación y las decisiones
Servidor MCP: Expone herramientas mediante transporte HTTP Streamable
Arcade: Aloja herramientas con autenticación, gestión de tokens y despliegue
Herramientas: Funciones personalizadas que construyes con Arcade SDK

Requisitos previos

Cuenta de Arcade y clave API
Python 3.8+
Node.js 18+ (para OAP)
Acceso a LangGraph Platform

Construcción de herramientas personalizadas

Configuración del entorno

# Install uv package manager
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create virtual environment
uv venv --seed
source .venv/bin/activate

# Install Arcade CLI
pip install arcade-ai

Crear Toolkit

arcade new company_tools
cd company_tools

Estructura generada:

company_tools/
├── arcade_company_tools/
│   └── tools/
│       └── __init__.py
├── evals/
├── tests/
└── pyproject.toml

Herramientas sin autenticación

Crea arcade_company_tools/tools/internal_api.py:

from typing import Annotated
from arcade.sdk import tool
import httpx

@tool
async def get_customer_tier(
    customer_id: Annotated[str, "Customer unique identifier"],
) -> dict:
    """Retrieve customer tier from internal CRM."""

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://crm.company.com/api/customers/{customer_id}/tier",
            headers={"X-API-Key": "YOUR_KEY"},
            timeout=30.0
        )
        response.raise_for_status()
        data = response.json()

    return {
        "customer_id": customer_id,
        "tier": data.get("tier"),
        "status": data.get("status")
    }

@tool
async def calculate_discount(
    customer_id: Annotated[str, "Customer identifier"],
    order_amount: Annotated[float, "Order total in dollars"],
) -> dict:
    """Calculate discount based on customer tier."""

    customer_data = await get_customer_tier(customer_id)
    tier = customer_data.get("tier", "bronze")

    rates = {"bronze": 0.05, "silver": 0.10, "gold": 0.15, "platinum": 0.20}
    rate = rates.get(tier, 0)
    discount = order_amount * rate

    return {
        "customer_id": customer_id,
        "tier": tier,
        "original_amount": order_amount,
        "discount_rate": rate,
        "discount_amount": discount,
        "final_amount": order_amount - discount
    }

Herramientas con OAuth

Crea arcade_company_tools/tools/salesforce.py:

from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Salesforce
from arcade.sdk.errors import RetryableToolError
import httpx

@tool(
    requires_auth=Salesforce(scopes=["api", "refresh_token"])
)
async def create_opportunity(
    context: ToolContext,
    account_name: Annotated[str, "Account name"],
    opportunity_name: Annotated[str, "Opportunity name"],
    amount: Annotated[float, "Deal amount"],
    close_date: Annotated[str, "Close date YYYY-MM-DD"],
) -> dict:
    """Create Salesforce opportunity."""

    if not context.authorization or not context.authorization.token:
        raise RetryableToolError(
            "Salesforce authorization required",
            developer_message="User must complete OAuth"
        )

    headers = {
        "Authorization": f"Bearer {context.authorization.token}",
        "Content-Type": "application/json"
    }

    instance_url = context.authorization.metadata.get("instance_url")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{instance_url}/services/data/v57.0/sobjects/Opportunity",
            headers=headers,
            json={
                "Name": opportunity_name,
                "AccountId": account_name,
                "Amount": amount,
                "CloseDate": close_date,
                "StageName": "Prospecting"
            },
            timeout=30.0
        )
        response.raise_for_status()

    return response.json()

@tool(
    requires_auth=Salesforce(scopes=["api", "refresh_token"])
)
async def search_contacts(
    context: ToolContext,
    email: Annotated[str, "Contact email address"],
) -> dict:
    """Search Salesforce contacts by email."""

    if not context.authorization or not context.authorization.token:
        raise RetryableToolError("Salesforce authorization required")

    headers = {
        "Authorization": f"Bearer {context.authorization.token}",
        "Content-Type": "application/json"
    }

    instance_url = context.authorization.metadata.get("instance_url")
    query = f"SELECT Id, Name, Email, Phone FROM Contact WHERE Email = '{email}'"

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{instance_url}/services/data/v57.0/query",
            headers=headers,
            params={"q": query},
            timeout=30.0
        )
        response.raise_for_status()

    return response.json()

Proveedores OAuth personalizados

Para servicios OAuth no estándar, crea arcade_company_tools/tools/custom_oauth.py:

from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import OAuth2
import httpx

@tool(
    requires_auth=OAuth2(
        id="internal_erp",
        provider_id="oauth2",
        scopes=["read:orders", "write:orders"]
    )
)
async def get_order_status(
    context: ToolContext,
    order_id: Annotated[str, "Order identifier"],
) -> dict:
    """Get order status from ERP."""

    if not context.authorization or not context.authorization.token:
        raise RetryableToolError("ERP authorization required")

    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://erp.company.com/api/orders/{order_id}",
            headers={"Authorization": f"Bearer {context.authorization.token}"},
            timeout=30.0
        )
        response.raise_for_status()

    return response.json()

, configura el proveedor OAuth en tu configuración de Arcade Engine o en el dashboard.

Gestionar secretos

Para claves API, usa los secretos de Arcade:

from arcade.sdk import tool, ToolContext
from arcade.sdk.auth import Secret
import httpx

@tool(
    requires_auth=Secret(
        key="stripe_api_key",
        description="Stripe API key"
    )
)
async def create_payment_intent(
    context: ToolContext,
    amount: Annotated[int, "Amount in cents"],
    currency: Annotated[str, "Currency code"],
) -> dict:
    """Create Stripe payment intent."""

    api_key = context.authorization.token

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.stripe.com/v1/payment_intents",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/x-www-form-urlencoded"
            },
            data={"amount": amount, "currency": currency},
            timeout=30.0
        )
        response.raise_for_status()

    return response.json()

Configura secretos en el dashboard de Arcade por usuario sin exponer el código.

Pruebas locales

Iniciar servidor de desarrollo

# Install dependencies
make install

# Start worker with auto-reload
arcade worker start --reload

Probar con Arcade Chat

arcade chat

Prueba las herramientas de forma interactiva:

You: Calculate discount for customer cust_123 with $500 order
Assistant: [Executes tool]
Result: Bronze tier, $25 discount, final $475

Crear evaluaciones

Agrega a evals/eval_company_tools.py:

from arcade.sdk.eval import EvalSuite, tool_eval

suite = EvalSuite(
    name="Company Tools",
    system_message="You handle customer operations.",
)

@tool_eval(tools=["CompanyTools.CalculateDiscount"])
async def test_discount():
    """Test discount calculation."""
    return [
        {
            "input": "Calculate discount for gold customer, $1000 order",
            "expected": "discount_amount: 150.0, final_amount: 850.0",
        },
        {
            "input": "Calculate discount for bronze customer, $500 order",
            "expected": "discount_amount: 25.0, final_amount: 475.0",
        },
    ]

, ejecuta las pruebas:

arcade evals run evals/eval_company_tools.py

Desplegar servidor MCP

Desplegar en Arcade Cloud

# Login
arcade login

# Deploy toolkit as MCP server
arcade deploy

URL de tu servidor MCP: https://api.arcade.dev/v1/mcps/YOUR_TOOLKIT/mcp

Características:

Escalado automático
Gestión de OAuth y tokens
Monitoreo y registro
Soporte para cloud y VPC

Servidor MCP auto-hospedado

Instalar Arcade Engine:

# macOS
brew install arcadeai/tap/arcade-engine

# Docker
docker pull ghcr.io/arcadeai/engine:latest

Crea engine.yaml:

auth:
  providers:
    - id: salesforce-provider
      enabled: true
      type: oauth2
      provider_id: salesforce
      client_id: ${env:SALESFORCE_CLIENT_ID}
      client_secret: ${env:SALESFORCE_CLIENT_SECRET}

    - id: internal-erp
      enabled: true
      type: oauth2
      provider_id: oauth2
      authorization_url: https://erp.company.com/oauth/authorize
      token_url: https://erp.company.com/oauth/token
      client_id: ${env:ERP_CLIENT_ID}
      client_secret: ${env:ERP_CLIENT_SECRET}

api:
  host: 0.0.0.0
  port: 9099

workers:
  - id: 'company-tools-worker'
    enabled: true
    http:
      uri: 'http://localhost:8002'
      secret: ${env:WORKER_SECRET}

, inicia los servicios:

# Terminal 1: Engine
arcade-engine start

# Terminal 2: Worker
arcade worker start

Servidor MCP disponible en: http://localhost:9099/v1/mcps/company_tools/mcp

Configurar Open Agent Platform

Establecer URL del servidor MCP

Agrega a apps/web/.env:

# Arcade Cloud
NEXT_PUBLIC_MCP_SERVER_URL="https://api.arcade.dev"
NEXT_PUBLIC_MCP_AUTH_REQUIRED=true

# Self-hosted
# NEXT_PUBLIC_MCP_SERVER_URL="http://localhost:9099"
# NEXT_PUBLIC_MCP_AUTH_REQUIRED=true

, actualiza los agentes existentes

Al cambiar la URL de MCP:

cd apps/web

# Set new URL
export NEXT_PUBLIC_MCP_SERVER_URL="https://api.arcade.dev"

# Update agents
npx tsx scripts/update-agents-mcp-url.ts

Esto actualiza todos los agentes desplegados con la nueva configuración MCP.

Crear agente de herramientas en OAP

Mediante la interfaz web de OAP:

Haz clic en “Crear agente”
Selecciona “Agente de herramientas”
Elige herramientas personalizadas de la lista del servidor MCP
Establece instrucciones del agente:

You handle customer support with CRM and order tools.

Capabilities:
- Customer information and tier lookup
- Discount calculations
- Order status checks
- Salesforce opportunity creation

Verify customer identity before accessing data.

Configura OAuth para las herramientas requeridas
Prueba en la interfaz de chat

Construir agente LangGraph con MCP

Creación programática de agentes:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

async def create_company_agent():
    """Create agent with company tools via MCP."""

    client = MultiServerMCPClient({
        "company_tools": {
            "url": "https://api.arcade.dev/v1/mcps/company_tools/mcp",
            "transport": "streamable_http",
        }
    })

    tools = await client.get_tools()
    model = ChatOpenAI(model="gpt-4o")
    agent = create_react_agent(model=model, tools=tools)

    return agent

# Use agent
agent = await create_company_agent()

response = await agent.ainvoke({
    "messages": [{
        "role": "user",
        "content": "Calculate discount for customer cust_123, $750 order"
    }]
})

print(response["messages"][-1].content)

Implementación de agente en JavaScript

import { MultiServerMCPClient } from 'langchain-mcp-adapters';
import { createReactAgent } from '@langchain/langgraph/prebuilt';
import { ChatOpenAI } from '@langchain/openai';

async function createCompanyAgent() {
  const client = new MultiServerMCPClient({
    company_tools: {
      url: 'https://api.arcade.dev/v1/mcps/company_tools/mcp',
      transport: 'streamable_http',
    },
  });

  const tools = await client.getTools();
  const model = new ChatOpenAI({ model: 'gpt-4o' });

  const agent = createReactAgent({
    llm: model,
    tools,
  });

  return agent;
}

const agent = await createCompanyAgent();

const result = await agent.invoke({
  messages: [
    {
      role: 'user',
      content: 'Look up tier for customer cust_456',
    },
  ],
});

console.log(result.messages[result.messages.length - 1].content);

Múltiples servidores MCP

Combina varios toolkits:

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    # Custom company tools
    "company_tools": {
        "url": "https://api.arcade.dev/v1/mcps/company_tools/mcp",
        "transport": "streamable_http",
    },
    # Pre-built Gmail toolkit
    "gmail": {
        "url": "https://api.arcade.dev/v1/mcps/arcade-anon/mcp",
        "transport": "streamable_http",
    },
})

tools = await client.get_tools()
agent = create_react_agent(ChatOpenAI(model="gpt-4o"), tools)

Gestionar autenticación

Flujo de autorización OAuth

Cuando las herramientas requieren OAuth:

Arcade verifica el estado de autorización del usuario
Devuelve la URL de autorización si es necesario
El usuario completa el OAuth en el navegador
Arcade almacena tokens cifrados
Las llamadas posteriores usan las credenciales almacenadas

Autorización en código

from arcadepy import AsyncArcade

async def run_with_auth(user_id: str, query: str):
    """Run agent with authorization handling."""

    client = AsyncArcade()

    auth_status = await client.tools.authorize(
        tool_name="CompanyTools.CreateSalesforceOpportunity",
        user_id=user_id
    )

    if auth_status.status != "completed":
        return {
            "requires_auth": True,
            "auth_url": auth_status.url,
            "message": "Authorize Salesforce access"
        }

    agent = await create_company_agent()

    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": query}],
        "configurable": {"user_id": user_id}
    })

    return result

Configurar autenticación MCP

Configuración de producción en apps/web/.env:

NEXT_PUBLIC_MCP_AUTH_REQUIRED=true
NEXT_PUBLIC_SUPABASE_URL="https://project.supabase.co"
NEXT_PUBLIC_SUPABASE_ANON_KEY="your-key"

, OAP intercambia JWTs de Supabase por tokens de acceso MCP.

Patrones avanzados

Coordinación multi-agente

Construye agentes supervisores:

# Sales agent
sales_mcp = MultiServerMCPClient({
    "company_tools": {
        "url": "https://api.arcade.dev/v1/mcps/company_tools/mcp",
        "transport": "streamable_http",
    }
})
sales_tools = await sales_mcp.get_tools()
sales_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    sales_tools,
    name="sales_agent"
)

# Support agent
support_mcp = MultiServerMCPClient({
    "company_tools": {
        "url": "https://api.arcade.dev/v1/mcps/company_tools/mcp",
        "transport": "streamable_http",
    },
    "gmail": {
        "url": "https://api.arcade.dev/v1/mcps/arcade-anon/mcp",
        "transport": "streamable_http",
    }
})
support_tools = await support_mcp.get_tools()
support_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o"),
    support_tools,
    name="support_agent"
)

# Supervisor
supervisor = create_supervisor_agent(
    agents=[sales_agent, support_agent],
    model=ChatOpenAI(model="gpt-4o")
)

Carga dinámica de herramientas

async def create_agent_with_tools(toolkits: list[str]):
    """Create agent with selected toolkits."""

    mcp_config = {
        toolkit: {
            "url": f"https://api.arcade.dev/v1/mcps/{toolkit}/mcp",
            "transport": "streamable_http",
        }
        for toolkit in toolkits
    }

    client = MultiServerMCPClient(mcp_config)
    tools = await client.get_tools()

    return create_react_agent(ChatOpenAI(model="gpt-4o"), tools)

# Specialized agents
sales_agent = await create_agent_with_tools(["company_tools"])
support_agent = await create_agent_with_tools(["company_tools", "gmail"])

Manejo de errores

Implementa lógica de reintentos:

from arcade.sdk.errors import RetryableToolError, ToolExecutionError
import asyncio

@tool
async def api_call_with_retry(
    context: ToolContext,
    endpoint: str,
) -> dict:
    """API call with retry logic."""

    max_retries = 3

    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    f"https://api.example.com/{endpoint}",
                    timeout=30.0
                )
                response.raise_for_status()
                return response.json()

        except httpx.TimeoutException:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            raise RetryableToolError(
                "Request timed out",
                developer_message=f"Timeout after {max_retries} retries"
            )

        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                raise RetryableToolError(
                    "Rate limit exceeded",
                    developer_message=e.response.text
                )
            raise ToolExecutionError(
                f"API error: {e.response.status_code}",
                developer_message=e.response.text
            )

Buenas prácticas en producción

Diseño de herramientas

Propósito único: Una acción por herramienta
Nombres claros: Usa verbos de acción como CreateInvoice, GetCustomerTier
Validar entradas: Sanitiza todos los parámetros antes de llamadas API
Retornos estructurados: Usa formatos de diccionario consistentes

Rendimiento

Operaciones asíncronas: Todas las herramientas deben ser async
Establecer timeouts: Usa límites de 30-60 segundos para solicitudes HTTP
Caché de datos: Reduce llamadas API para datos consultados frecuentemente

_cache = {}
_cache_ttl = 300

@tool
async def get_cached_catalog() -> dict:
    """Get product catalog with caching."""

    now = time.time()

    if "catalog" in _cache:
        cached_data, timestamp = _cache["catalog"]
        if now - timestamp < _cache_ttl:
            return cached_data

    data = await fetch_catalog()
    _cache["catalog"] = (data, now)
    return data

Seguridad

Sin registrar secretos: Nunca registres tokens ni claves API
Mínimo privilegio: Solicita solo los scopes OAuth necesarios
Verificar autorización: Verifica que context.authorization existe
Sanitizar salidas: Elimina datos sensibles de las respuestas
Secretos de entorno: Usa variables de entorno, no valores hardcodeados

Monitoreo

import logging
from datetime import datetime

logger = logging.getLogger(__name__)

@tool
async def monitored_tool(context: ToolContext, param: str) -> dict:
    """Tool with monitoring."""

    start = datetime.now()

    try:
        result = await perform_operation(param)

        duration = (datetime.now() - start).total_seconds()
        logger.info(
            "Tool executed",
            extra={
                "tool": "monitored_tool",
                "duration": duration,
                "user_id": context.user_id
            }
        )
        return result

    except Exception as e:
        logger.error(
            "Tool failed",
            extra={
                "tool": "monitored_tool",
                "error": str(e),
                "user_id": context.user_id
            }
        )
        raise

Solución de problemas

Falla de conexión MCP

Problema: OAP no puede conectarse al servidor MCP

Solución:

Verifica la URL: curl https://api.arcade.dev/v1/mcps/company_tools/mcp
Asegúrate de que NEXT_PUBLIC_MCP_SERVER_URL excluya el /mcp sufijo
Verifica que el servidor MCP esté activo

Herramientas no visibles

Problema: Herramientas personalizadas ausentes del agente

Solución:

Verifica el despliegue: arcade show
Reinicia la aplicación web de OAP
Verifica que el nombre del toolkit coincida exactamente con la URL MCP

Bucles de autorización

Problema: La autorización OAuth falla repetidamente

Solución:

Verifica la configuración del proveedor OAuth en Arcade
Revisa que las URLs de redirección coincidan en la app OAuth
Asegúrate de que el usuario completó el flujo OAuth completo
Limpia el caché del navegador

Timeouts de herramientas

Problema: Las herramientas agotan el tiempo durante la ejecución

Solución:

Aumenta los valores de timeout
Revisa el rendimiento del endpoint API
Agrega lógica de reintentos para fallos transitorios
Divide las operaciones en herramientas más pequeñas

Recursos

Resumen

Construye herramientas personalizadas con Arcade SDK, despliégalas como servidores MCP e intégralas con la Open Agent Platform de LangChain. Arcade gestiona la autenticación, los tokens y la infraestructura mientras tú te enfocas en la funcionalidad de las herramientas.

El protocolo MCP crea integraciones reutilizables que funcionan en distintas plataformas de agentes. Construye una vez con Arcade, despliega en la nube o auto-hospeda, y úsalo en cualquier parte.

Empieza a construir en arcade.dev.

Cómo llamar herramientas personalizadas del Open Agents SDK via Arcade.dev

Resumen de arquitectura

Requisitos previos

Construcción de herramientas personalizadas

Configuración del entorno

Crear Toolkit

Herramientas sin autenticación

Herramientas con OAuth

Proveedores OAuth personalizados

Gestionar secretos

Pruebas locales

Iniciar servidor de desarrollo

Probar con Arcade Chat

Crear evaluaciones

Desplegar servidor MCP

Desplegar en Arcade Cloud

Servidor MCP auto-hospedado

Configurar Open Agent Platform

Establecer URL del servidor MCP

, actualiza los agentes existentes

Crear agente de herramientas en OAP

Construir agente LangGraph con MCP

Implementación de agente en JavaScript

Múltiples servidores MCP

Gestionar autenticación

Flujo de autorización OAuth

Autorización en código

Configurar autenticación MCP

Patrones avanzados

Coordinación multi-agente

Carga dinámica de herramientas

Manejo de errores

Buenas prácticas en producción

Diseño de herramientas

Rendimiento

Seguridad

Monitoreo

Solución de problemas

Falla de conexión MCP

Herramientas no visibles

Bucles de autorización

Timeouts de herramientas

Recursos

Resumen

ARTÍCULOS RECIENTES

Welcome to Arcade Labs

Microsoft Build 2026: What Was Announced, and What It Means for the Agent Stack

Opus 4.8 just dropped. But can it actually do the work?

Ingresa a Arcade ahora y comienza a construir.