How to Call Custom Tools from Open Agents SDK via Arcade

LangChain's Open Agent Platform connects to tools through the Model Context Protocol (MCP). Arcade provides the infrastructure to build, deploy, and serve custom tools as MCP servers that OAP agents can call.

This guide covers building custom tools with Arcade's SDK, deploying them as MCP servers, and integrating them into LangChain Open Agent Platform.

Architecture Overview

The integration flow works as follows:

LangGraph Agent → MCP Protocol → Arcade MCP Server → Custom Tool Execution → Response

Components:

LangGraph Agents: Run on LangGraph Platform, handle conversation and decisions
MCP Server: Exposes tools via HTTP Streamable transport
Arcade: Hosts tools with authentication, token management, and deployment
Tools: Custom functions you build with Arcade SDK

Prerequisites

Arcade account and API key
Python 3.8+
Node.js 18+ (for OAP)
LangGraph Platform access

Building Custom Tools

Environment Setup

# 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

Create Toolkit

arcade new company_tools
cd company_tools

Structure created:

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

Build Tools Without Authentication

Create 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
    }

Build Tools With OAuth

Create 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()

Custom OAuth Providers

For non-standard OAuth services, create 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()

Configure the OAuth provider in your Arcade Engine configuration or dashboard.

Manage Secrets

For API keys, use Arcade's secrets:

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()

Configure secrets in the Arcade dashboard per user without code exposure.

Local Testing

Start Development Server

# Install dependencies
make install

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

Test with Arcade Chat

arcade chat

Test tools interactively:

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

Create Evaluations

Add to 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",
        },
    ]

Run tests:

arcade evals run evals/eval_company_tools.py

Deploy MCP Server

Deploy to Arcade Cloud

# Login
arcade login

# Deploy toolkit as MCP server
arcade deploy

Your MCP server URL: https://api.arcade.dev/v1/mcps/YOUR_TOOLKIT/mcp

Features:

Automatic scaling
OAuth and token management
Monitoring and logging
Cloud and VPC support

Self-Host MCP Server

Install Arcade Engine:

# macOS
brew install arcadeai/tap/arcade-engine

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

Create 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}

Start services:

# Terminal 1: Engine
arcade-engine start

# Terminal 2: Worker
arcade worker start

MCP server available at: http://localhost:9099/v1/mcps/company_tools/mcp

Configure Open Agent Platform

Set MCP Server URL

Add to 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

Update Existing Agents

When changing MCP URL:

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

This updates all deployed agents with the new MCP configuration.

Create Tools Agent in OAP

Via OAP web interface:

Click "Create Agent"
Select "Tools Agent"
Choose custom tools from MCP server list
Set agent instructions:

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.

Configure OAuth for required tools
Test in chat interface

Build LangGraph Agent with MCP

Programmatic agent creation:

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)

JavaScript Agent Implementation

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);

Multiple MCP Servers

Combine multiple 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)

Handle Authentication

OAuth Authorization Flow

When tools require OAuth:

Arcade checks user authorization status
Returns authorization URL if needed
User completes OAuth in browser
Arcade stores encrypted tokens
Subsequent calls use stored credentials

Authorization in 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

Configure MCP Authentication

Production configuration in 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 exchanges Supabase JWTs for MCP access tokens.

Advanced Patterns

Multi-Agent Coordination

Build supervisor agents:

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

Dynamic Tool Loading

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

Error Handling

Implement retry logic:

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
            )

Production Best Practices

Tool Design

Single purpose: One action per tool
Clear names: Use action verbs like CreateInvoice, GetCustomerTier
Validate inputs: Sanitize all parameters before API calls
Structured returns: Use consistent dictionary formats

Performance

Async operations: All tools must be async
Set timeouts: Use 30-60 second limits for HTTP requests
Cache data: Reduce API calls for frequently accessed data

_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

Security

No logging of secrets: Never log tokens or API keys
Least privilege: Request minimum required OAuth scopes
Check authorization: Verify context.authorization exists
Sanitize outputs: Remove sensitive data from responses
Environment secrets: Use environment variables, not hardcoded values

Monitoring

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

Troubleshooting

MCP Connection Fails

Issue: OAP cannot connect to MCP server

Fix:

Verify URL: curl https://api.arcade.dev/v1/mcps/company_tools/mcp
Ensure NEXT_PUBLIC_MCP_SERVER_URL excludes /mcp suffix
Check MCP server is running

Tools Not Visible

Issue: Custom tools missing from agent

Fix:

Verify deployment: arcade show
Restart OAP web application
Check toolkit name matches MCP URL exactly

Authorization Loops

Issue: OAuth authorization fails repeatedly

Fix:

Verify OAuth provider configuration in Arcade
Check redirect URLs match in OAuth app
Ensure user completed full OAuth flow
Clear browser cache

Tool Timeouts

Issue: Tools timeout during execution

Fix:

Increase timeout values
Check API endpoint performance
Add retry logic for transient failures
Split operations into smaller tools

Resources

Summary

Build custom tools with Arcade SDK, deploy them as MCP servers, and integrate with LangChain Open Agent Platform. Arcade handles authentication, token management, and infrastructure while you focus on tool functionality.

The MCP protocol creates reusable integrations that work across agent platforms. Build once with Arcade, deploy to cloud or self-host, and use anywhere.

Start building at arcade.dev.

How to Call Custom Tools from Open Agents SDK via Arcade

Architecture Overview

Prerequisites

Building Custom Tools

Environment Setup

Create Toolkit

Build Tools Without Authentication

Build Tools With OAuth

Custom OAuth Providers

Manage Secrets

Local Testing

Start Development Server

Test with Arcade Chat

Create Evaluations

Deploy MCP Server

Deploy to Arcade Cloud

Self-Host MCP Server

Configure Open Agent Platform

Set MCP Server URL

Update Existing Agents

Create Tools Agent in OAP

Build LangGraph Agent with MCP

JavaScript Agent Implementation

Multiple MCP Servers

Handle Authentication

OAuth Authorization Flow

Authorization in Code

Configure MCP Authentication

Advanced Patterns

Multi-Agent Coordination

Dynamic Tool Loading

Error Handling

Production Best Practices

Tool Design

Performance

Security

Monitoring

Troubleshooting

MCP Connection Fails

Tools Not Visible

Authorization Loops

Tool Timeouts

Resources

Summary

RECENT ARTICLES

How to Query Postgres from Python Agent via Arcade (MCP)

How to Connect Python Agent to Slack with Arcade (MCP)

How to Build a Python Gmail Agent with Arcade (MCP)

Get early access to Arcade, and start building now.