Appeler des outils personnalisés via Arcade.dev

La plateforme Open Agent de LangChain se connecte aux outils via le Model Context Protocol (MCP). Arcade.dev fournit l’infrastructure pour construire, déployer et exposer des outils personnalisés sous forme de serveurs MCP que les agents OAP peuvent appeler.

Ce guide explique comment créer des outils personnalisés avec le SDK Arcade, les déployer en tant que serveurs MCP et les intégrer dans la plateforme LangChain Open Agent.

Aperçu de l’architecture

Le flux d’intégration fonctionne comme suit :

Agent LangGraph → Protocole MCP → Serveur MCP Arcade → Exécution de l’outil personnalisé → Réponse

Composants :

Agents LangGraph : Tournent sur LangGraph Platform, gèrent la conversation et les prises de décision
Serveur MCP : Expose les outils via le transport HTTP Streamable
Arcade : Héberge les outils avec authentification, gestion des tokens et déploiement
Outils : Fonctions personnalisées que vous créez avec Arcade SDK

Prérequis

Compte Arcade et clé API
Python 3.8+
Node.js 18+ (pour OAP)
Accès à LangGraph Platform

Créer des outils personnalisés

Configuration de l’environnement

# 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

Créer un toolkit

arcade new company_tools
cd company_tools

Structure créée :

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

Créer des outils sans authentification

Créer 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
    }

Créer des outils avec OAuth

Créer 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()

Fournisseurs OAuth personnalisés

Pour les services OAuth non standard, créez 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()

Configurez le fournisseur OAuth dans votre configuration Arcade Engine ou le tableau de bord.

Gérer les secrets

Pour les clés API, utilisez les secrets d’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()

Configurez les secrets dans le tableau de bord Arcade par utilisateur, sans exposer le code.

Tests en local

Démarrer le serveur de développement

# Install dependencies
make install

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

Tester avec Arcade Chat

arcade chat

Testez les outils de façon interactive :

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

Créer des évaluations

Ajouter à 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",
        },
    ]

Lancer les tests :

arcade evals run evals/eval_company_tools.py

Déployer le serveur MCP

Déployer sur Arcade Cloud

# Login
arcade login

# Deploy toolkit as MCP server
arcade deploy

URL de votre serveur MCP : https://api.arcade.dev/v1/mcps/YOUR_TOOLKIT/mcp

Fonctionnalités :

Mise à l’échelle automatique
Gestion OAuth et des tokens
Supervision et journalisation
Support cloud et VPC

Auto-héberger le serveur MCP

Installer Arcade Engine :

# macOS
brew install arcadeai/tap/arcade-engine

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

Créer 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}

Démarrer les services :

# Terminal 1: Engine
arcade-engine start

# Terminal 2: Worker
arcade worker start

Serveur MCP disponible à l’adresse : http://localhost:9099/v1/mcps/company_tools/mcp

Configurer Open Agent Platform

Définir l’URL du serveur MCP

Ajouter à 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

Mettre à jour les agents existants

Lors du changement d’URL 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

Cela met à jour tous les agents déployés avec la nouvelle configuration MCP.

Créer un agent Tools dans OAP

Via l’interface web OAP :

Cliquez sur « Create Agent »
Sélectionnez « Tools Agent »
Choisissez des outils personnalisés depuis la liste des serveurs MCP
Définissez les instructions de l’agent :

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.

Configurez OAuth pour les outils requis
Testez dans l’interface de chat

Créer un agent LangGraph avec MCP

Création d’agent par programmation :

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)

Implémentation d’agent 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);

Plusieurs serveurs MCP

Combinez plusieurs 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)

Gérer l’authentification

Flux d’autorisation OAuth

Quand les outils nécessitent OAuth :

Arcade vérifie le statut d’autorisation de l’utilisateur
Retourne l’URL d’autorisation si nécessaire
L’utilisateur complète OAuth dans le navigateur
Arcade stocke les tokens chiffrés
Les appels suivants utilisent les identifiants stockés

Autorisation dans le code

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

Configurer l’authentification MCP

Configuration de production dans 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 échange les JWT Supabase contre des tokens d’accès MCP.

Modèles avancés

Coordination multi-agents

Créez des agents superviseurs :

# 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")
)

Chargement dynamique des outils

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"])

Gestion des erreurs

Implémentez une logique de retry :

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
            )

Bonnes pratiques en production

Conception des outils

Usage unique : Une action par outil
Noms explicites : Utilisez des verbes d’action comme CreateInvoice, GetCustomerTier
Valider les entrées : Assainissez tous les paramètres avant les appels API
Retours structurés : Utilisez des formats de dictionnaire cohérents

Performance

Opérations asynchrones : Tous les outils doivent être asynchrones
Définir des délais d’expiration : Limitez les requêtes HTTP à 30-60 secondes
Mettre en cache les données : Réduisez les appels API pour les données fréquemment consultées

_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

Sécurité

Pas de journalisation des secrets : Ne journalisez jamais les tokens ni les clés API
Moindre privilège : Demandez le minimum de scopes OAuth nécessaires
Vérifier les autorisations : Vérifiez que context.authorization existe
Assainir les sorties : Supprimez les données sensibles des réponses
Secrets d’environnement : Utilisez des variables d’environnement, pas de valeurs codées en dur

Surveillance

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

Dépannage

Échec de connexion MCP

Problème : OAP ne peut pas se connecter au serveur MCP

Correction :

Vérifiez l’URL : curl https://api.arcade.dev/v1/mcps/company_tools/mcp
Assurez-vous que NEXT_PUBLIC_MCP_SERVER_URL exclut le suffixe /mcp
Vérifiez que le serveur MCP est en cours d’exécution

Outils non visibles

Problème : Outils personnalisés absents de l’agent

Correctif :

Vérifier le déploiement : arcade show
Redémarrer l’application web OAP
Vérifier que le nom du toolkit correspond exactement à l’URL MCP

Boucles d’autorisation

Problème : l’autorisation OAuth échoue de manière répétée

Correctif :

Vérifier la configuration du fournisseur OAuth dans Arcade
Vérifier que les URL de redirection correspondent dans l’application OAuth
S’assurer que l’utilisateur a complété le flux OAuth
Vider le cache du navigateur

Délais d’expiration des outils

Problème : les outils expirent pendant l’exécution

Correctif :

Augmenter les valeurs de délai d’expiration
Vérifier les performances de l’endpoint API
Ajouter une logique de réessai pour les erreurs transitoires
Diviser les opérations en outils plus petits

Ressources

Résumé

Créez des outils personnalisés avec Arcade SDK, déployez-les comme serveurs MCP et intégrez-les à LangChain Open Agent Platform. Arcade gère l’authentification, les tokens et l’infrastructure pendant que vous vous concentrez sur les fonctionnalités.

Le protocole MCP crée des intégrations réutilisables qui fonctionnent sur différentes plateformes d’agents. Créez une fois avec Arcade, déployez dans le cloud ou en auto-hébergement, et utilisez partout.

Commencez à construire sur arcade.dev.

Comment appeler des outils personnalisés depuis l'Open Agents SDK via Arcade.dev

Aperçu de l’architecture

Prérequis

Créer des outils personnalisés

Configuration de l’environnement

Créer un toolkit

Créer des outils sans authentification

Créer des outils avec OAuth

Fournisseurs OAuth personnalisés

Gérer les secrets

Tests en local

Démarrer le serveur de développement

Tester avec Arcade Chat

Créer des évaluations

Déployer le serveur MCP

Déployer sur Arcade Cloud

Auto-héberger le serveur MCP

Configurer Open Agent Platform

Définir l’URL du serveur MCP

Mettre à jour les agents existants

Créer un agent Tools dans OAP

Créer un agent LangGraph avec MCP

Implémentation d’agent JavaScript

Plusieurs serveurs MCP

Gérer l’authentification

Flux d’autorisation OAuth

Autorisation dans le code

Configurer l’authentification MCP

Modèles avancés

Coordination multi-agents

Chargement dynamique des outils

Gestion des erreurs

Bonnes pratiques en production

Conception des outils

Performance

Sécurité

Surveillance

Dépannage

Échec de connexion MCP

Outils non visibles

Boucles d’autorisation

Délais d’expiration des outils

Ressources

Résumé

RECENT ARTICLES

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?

Accédez en avant-première à Arcade et commencez à construire dès maintenant.