Custom MCP tools

Overview

Extend MagOneAI with custom tools for your internal systems, proprietary APIs, and unique business logic. Any service with an API can become an MCP tool that your agents can use. Custom tools enable agents to interact with:

Internal APIs — your company’s microservices and internal tools
Proprietary systems — ERP, CRM, inventory management, etc.
Third-party services — APIs not natively integrated with MagOneAI
Legacy systems — wrap older SOAP or REST APIs with MCP interface
Custom business logic — implement domain-specific operations as tools

The key advantage: you build it once using the open MCP standard, and it works with MagOneAI and any other MCP-compatible AI platform.

How to add custom MCP tools

Follow this workflow to add custom tools to your MagOneAI projects.

Build or deploy an MCP server

Create an MCP server that wraps your API. You can:

Build from scratch using MCP SDKs (Python or TypeScript)
Use an existing MCP server from the community
Deploy pre-built MCP servers for common services

The MCP server runs as a separate process and exposes tools via the MCP protocol.

Define tool schemas

Each tool needs:

Name: Unique identifier (e.g., get_customer_details)
Description: What the tool does (used by agents to decide when to use it)
Parameters: JSON Schema defining inputs
Return type: Schema for response structure

The agent uses these definitions to understand how to call your tools.

In Project Settings → Integrations → Custom Tools, add your MCP server:

Server URL (e.g., http://mcp-server:8000)
Connection protocol (HTTP, WebSocket, or stdio)
Authentication details (if required)

Configure authentication

Provide credentials the MCP server needs to access your API:

API keys
OAuth tokens
Basic auth credentials
Custom headers

Credentials are stored in HashiCorp Vault and injected at runtime.

Test the tool connection

Use the built-in tool tester to verify:

MCP server is reachable
Tool definitions are valid
Test calls execute successfully
Authentication works correctly

Attach the tool to agents

Enable the custom tool in agent configurations. Agents can now discover and use your custom tools.

MCP server development basics

MCP servers are lightweight processes that expose tools to AI agents. They implement the Model Context Protocol specification.

Architecture

The MCP server acts as an adapter between the standardized MCP protocol and your specific API or system.

Available SDKs

Build MCP servers using official SDKs:

Python SDK

pip install mcp

Best for: Data processing, ML models, database integrations

TypeScript SDK

npm install @modelcontextprotocol/sdk

Best for: Web APIs, real-time services, Node.js ecosystems

Basic MCP server structure (Python)

Here’s a minimal MCP server that wraps a REST API:

from mcp import Server, Tool
import httpx

# Initialize MCP server
server = Server("my-crm-integration")

# Define a tool
@server.tool(
    name="search_contacts",
    description="Search for contacts in the CRM by name, email, or company",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search query"
            },
            "limit": {
                "type": "integer",
                "description": "Max results to return",
                "default": 10
            }
        },
        "required": ["query"]
    }
)
async def search_contacts(query: str, limit: int = 10):
    # Call your CRM API
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.yourcrm.com/contacts/search",
            params={"q": query, "limit": limit},
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        return response.json()

# Run the server
if __name__ == "__main__":
    server.run()

Basic MCP server structure (TypeScript)

The same server in TypeScript:

import { Server } from '@modelcontextprotocol/sdk';
import axios from 'axios';

// Initialize MCP server
const server = new Server({ name: 'my-crm-integration' });

// Define a tool
server.addTool({
  name: 'search_contacts',
  description: 'Search for contacts in the CRM by name, email, or company',
  parameters: {
    type: 'object',
    properties: {
      query: {
        type: 'string',
        description: 'Search query',
      },
      limit: {
        type: 'integer',
        description: 'Max results to return',
        default: 10,
      },
    },
    required: ['query'],
  },
  execute: async ({ query, limit = 10 }) => {
    // Call your CRM API
    const response = await axios.get('https://api.yourcrm.com/contacts/search', {
      params: { q: query, limit },
      headers: { Authorization: `Bearer ${API_KEY}` },
    });
    return response.data;
  },
});

// Run the server
server.listen();

Tool definition best practices

Write clear, descriptive tool names

Tool names should be descriptive and follow a consistent naming convention.✅ Good: search_customers, create_support_ticket, get_inventory_status❌ Bad: search, create, get_dataAgents use tool names to understand what the tool does. Clear names improve agent decision-making.

Provide detailed descriptions

Descriptions help agents decide when to use a tool. Include:

What the tool does
What data it returns
When it should be used

✅ Good: “Search for customers in the CRM by name, email, or company. Returns customer ID, contact details, and account status. Use this to find customer information before creating support tickets.”❌ Bad: “Search customers”

Use JSON Schema for parameter validation

Define precise parameter schemas:

Required vs optional parameters
Data types (string, integer, boolean, array, object)
Constraints (min/max values, string patterns, enums)
Default values

MagOneAI validates parameters before execution, preventing invalid tool calls.

Return structured data

Always return structured JSON responses, not plain text. This allows agents to:

Extract specific fields
Pass data to subsequent tools
Format information appropriately

✅ Good:

{
  "customer_id": "12345",
  "name": "Acme Corp",
  "status": "active",
  "contacts": [...]
}

❌ Bad:

{
  "result": "Customer Acme Corp (ID: 12345) is active"
}

Handle errors gracefully

Return structured error responses that agents can interpret:

{
    "success": False,
    "error": "Customer not found",
    "error_code": "NOT_FOUND"
}

This allows agents to handle errors intelligently (e.g., “I couldn’t find that customer” instead of crashing).

Example: Custom CRM integration

Let’s build a complete MCP server for a fictional CRM system.

Tools to implement

Our CRM integration will expose these tools:

search_contacts — find contacts by name, email, or company
get_deal_status — retrieve status of a sales opportunity
update_opportunity — modify deal amount, stage, or close date
log_activity — record a call, email, or meeting with a contact

Complete Python implementation

from mcp import Server, Tool
from typing import Optional, List, Dict, Any
import httpx
import os

# Initialize server
server = Server("crm-integration")

# Get API credentials from environment
CRM_API_URL = os.getenv("CRM_API_URL", "https://api.crm.example.com")
CRM_API_KEY = os.getenv("CRM_API_KEY")

# Tool 1: Search contacts
@server.tool(
    name="search_contacts",
    description="Search for contacts in the CRM by name, email, or company name",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "limit": {"type": "integer", "default": 10, "description": "Max results"}
        },
        "required": ["query"]
    }
)
async def search_contacts(query: str, limit: int = 10) -> Dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CRM_API_URL}/contacts/search",
            params={"q": query, "limit": limit},
            headers={"Authorization": f"Bearer {CRM_API_KEY}"}
        )
        if response.status_code == 200:
            return {"success": True, "contacts": response.json()}
        else:
            return {"success": False, "error": "Search failed"}

# Tool 2: Get deal status
@server.tool(
    name="get_deal_status",
    description="Retrieve the current status of a sales opportunity by deal ID",
    parameters={
        "type": "object",
        "properties": {
            "deal_id": {"type": "string", "description": "Deal/opportunity ID"}
        },
        "required": ["deal_id"]
    }
)
async def get_deal_status(deal_id: str) -> Dict[str, Any]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{CRM_API_URL}/deals/{deal_id}",
            headers={"Authorization": f"Bearer {CRM_API_KEY}"}
        )
        if response.status_code == 200:
            deal = response.json()
            return {
                "success": True,
                "deal_id": deal["id"],
                "name": deal["name"],
                "stage": deal["stage"],
                "amount": deal["amount"],
                "close_date": deal["expected_close_date"]
            }
        elif response.status_code == 404:
            return {"success": False, "error": "Deal not found"}
        else:
            return {"success": False, "error": "Failed to retrieve deal"}

# Tool 3: Update opportunity
@server.tool(
    name="update_opportunity",
    description="Update a sales opportunity's amount, stage, or close date",
    parameters={
        "type": "object",
        "properties": {
            "deal_id": {"type": "string", "description": "Deal ID to update"},
            "amount": {"type": "number", "description": "New deal amount"},
            "stage": {"type": "string", "enum": ["prospecting", "qualification", "proposal", "negotiation", "closed_won", "closed_lost"]},
            "close_date": {"type": "string", "format": "date", "description": "Expected close date (YYYY-MM-DD)"}
        },
        "required": ["deal_id"]
    }
)
async def update_opportunity(deal_id: str, amount: Optional[float] = None,
                            stage: Optional[str] = None, close_date: Optional[str] = None) -> Dict[str, Any]:
    update_data = {}
    if amount is not None:
        update_data["amount"] = amount
    if stage is not None:
        update_data["stage"] = stage
    if close_date is not None:
        update_data["expected_close_date"] = close_date

    async with httpx.AsyncClient() as client:
        response = await client.patch(
            f"{CRM_API_URL}/deals/{deal_id}",
            json=update_data,
            headers={"Authorization": f"Bearer {CRM_API_KEY}"}
        )
        if response.status_code == 200:
            return {"success": True, "message": "Deal updated successfully"}
        else:
            return {"success": False, "error": "Failed to update deal"}

# Tool 4: Log activity
@server.tool(
    name="log_activity",
    description="Log an activity (call, email, meeting) with a contact",
    parameters={
        "type": "object",
        "properties": {
            "contact_id": {"type": "string", "description": "Contact ID"},
            "activity_type": {"type": "string", "enum": ["call", "email", "meeting"], "description": "Type of activity"},
            "subject": {"type": "string", "description": "Activity subject/title"},
            "notes": {"type": "string", "description": "Activity notes/description"},
            "date": {"type": "string", "format": "date-time", "description": "Activity date and time"}
        },
        "required": ["contact_id", "activity_type", "subject"]
    }
)
async def log_activity(contact_id: str, activity_type: str, subject: str,
                      notes: Optional[str] = None, date: Optional[str] = None) -> Dict[str, Any]:
    activity_data = {
        "contact_id": contact_id,
        "type": activity_type,
        "subject": subject,
        "notes": notes,
        "date": date or datetime.now().isoformat()
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{CRM_API_URL}/activities",
            json=activity_data,
            headers={"Authorization": f"Bearer {CRM_API_KEY}"}
        )
        if response.status_code == 201:
            return {"success": True, "activity_id": response.json()["id"]}
        else:
            return {"success": False, "error": "Failed to log activity"}

# Run the server
if __name__ == "__main__":
    server.run(host="0.0.0.0", port=8000)

Deploying the MCP server

Package and deploy your MCP server:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY crm_mcp_server.py .

CMD ["python", "crm_mcp_server.py"]

# Build and run
docker build -t crm-mcp-server .
docker run -p 8000:8000 \
  -e CRM_API_URL=https://api.crm.example.com \
  -e CRM_API_KEY=your_api_key \
  crm-mcp-server

Now register this server in MagOneAI at http://your-server:8000.

Authentication for custom integrations

Your MCP server needs credentials to access your APIs. MagOneAI provides secure credential management.

API key authentication

For services using API keys:

Store the API key in MagOneAI’s Vault (Admin Portal → Secrets → Add Secret)
Reference the secret when configuring the MCP server: vault:crm/api_key
MagOneAI injects the key into the MCP server environment at runtime

Your MCP server reads the key from environment variables:

API_KEY = os.getenv("CRM_API_KEY")  # Injected by MagOneAI

OAuth2 for custom services

For OAuth2-protected APIs, MagOneAI can manage the OAuth flow:

Configure OAuth app

Add OAuth config to MagOneAI

In Admin Portal, add OAuth configuration:

Client ID
Client secret
Authorization URL
Token URL
Required scopes

Users authorize access

When users first use your tool, they complete the OAuth flow. MagOneAI stores their access and refresh tokens in Vault.

Tokens injected at runtime

When an agent calls your tool, MagOneAI injects the user’s access token as an environment variable that your MCP server can use.

Custom authentication schemes

For custom authentication (HMAC signatures, JWT, etc.), implement authentication in your MCP server and store necessary secrets in Vault.

Testing custom tools

Before deploying to production, thoroughly test your custom tools.

Local testing

Test your MCP server locally using the MCP Inspector:

# Install MCP Inspector
npm install -g @modelcontextprotocol/inspector

# Run your MCP server
python crm_mcp_server.py

# In another terminal, run the inspector
mcp-inspector http://localhost:8000

The inspector provides a UI to:

View tool definitions
Execute tools with test parameters
Inspect request/response payloads
Debug connection issues

Integration testing in MagOneAI

Once registered in MagOneAI, use the built-in tool tester:

Go to Project Settings → Integrations → Custom Tools → Your Tool
Click Test Tool
Select a tool to test
Enter test parameters
Execute and verify the response

Testing in workflows

Create a test workflow that uses your custom tools:

Build a simple workflow with an Agent node
Attach your custom tools to the agent
Provide test prompts that should trigger tool usage
Verify the agent calls tools correctly and handles responses

Start with the MCP Python or TypeScript SDK. A basic MCP server that wraps a REST API can be built in under an hour. Begin with one or two tools, test thoroughly, then expand.

Common patterns and examples

Pattern: Database wrapper

Expose database access as MCP tools for agents that need data beyond standard SQL:

@server.tool(name="get_customer_lifetime_value", ...)
async def get_customer_ltv(customer_id: str):
    # Complex query with business logic
    result = await db.execute("""
        SELECT
            c.customer_id,
            c.name,
            COALESCE(SUM(o.total), 0) as lifetime_value,
            COUNT(o.id) as order_count
        FROM customers c
        LEFT JOIN orders o ON c.id = o.customer_id
        WHERE c.id = ? AND o.status = 'completed'
        GROUP BY c.id
    """, [customer_id])
    return result

Pattern: Multi-step workflow

Implement tools that orchestrate multiple API calls:

@server.tool(name="onboard_new_customer", ...)
async def onboard_customer(name: str, email: str, plan: str):
    # Step 1: Create customer in CRM
    customer = await crm_api.create_customer(name, email)

    # Step 2: Create subscription
    subscription = await billing_api.create_subscription(customer.id, plan)

    # Step 3: Send welcome email
    await email_api.send_welcome(email, customer.id)

    return {
        "customer_id": customer.id,
        "subscription_id": subscription.id,
        "status": "onboarded"
    }

Pattern: Webhook receiver

Accept webhooks and expose them as resources or events to agents:

# MCP server can receive webhooks and make data available
@server.resource(name="recent_support_tickets")
async def get_recent_tickets():
    # Return tickets received via webhook in last 24 hours
    return recent_tickets_cache

Agents can query this resource to stay informed about incoming events.

Next steps

MCP protocol

Deep dive into how MCP works

Tools overview

Learn about MagOneAI’s tool architecture

Security best practices

Secure your custom integrations

Workflow builder

Use custom tools in workflows

Getting Started

Platform Guide

Agents

Workflow Builder

Tools & Integrations

Models & Providers

Cookbooks

Security & Administration

Overview

How to add custom MCP tools

MCP server development basics

Architecture

Available SDKs

Python SDK

TypeScript SDK

Basic MCP server structure (Python)

Basic MCP server structure (TypeScript)

Tool definition best practices

Example: Custom CRM integration

Tools to implement

Complete Python implementation

Deploying the MCP server

Authentication for custom integrations

API key authentication

OAuth2 for custom services

Custom authentication schemes

Testing custom tools

Local testing

Integration testing in MagOneAI

Testing in workflows

Common patterns and examples

Pattern: Database wrapper

Pattern: Multi-step workflow

Pattern: Webhook receiver

Next steps

MCP protocol

Tools overview

Security best practices

Workflow builder

Getting Started

Platform Guide

Agents

Workflow Builder

Tools & Integrations

Models & Providers

Cookbooks

Security & Administration

​Overview

​How to add custom MCP tools

​MCP server development basics

​Architecture

​Available SDKs

Python SDK

TypeScript SDK

​Basic MCP server structure (Python)

​Basic MCP server structure (TypeScript)

​Tool definition best practices

​Example: Custom CRM integration

​Tools to implement

​Complete Python implementation

​Deploying the MCP server

​Authentication for custom integrations

​API key authentication

​OAuth2 for custom services

​Custom authentication schemes

​Testing custom tools

​Local testing

​Integration testing in MagOneAI

​Testing in workflows

​Common patterns and examples

​Pattern: Database wrapper

​Pattern: Multi-step workflow

​Pattern: Webhook receiver

​Next steps

MCP protocol

Tools overview

Security best practices

Workflow builder

Overview

How to add custom MCP tools

MCP server development basics

Architecture

Available SDKs

Basic MCP server structure (Python)

Basic MCP server structure (TypeScript)

Tool definition best practices

Example: Custom CRM integration

Tools to implement

Complete Python implementation

Deploying the MCP server

Authentication for custom integrations

API key authentication

OAuth2 for custom services

Custom authentication schemes

Testing custom tools

Local testing

Integration testing in MagOneAI

Testing in workflows

Common patterns and examples

Pattern: Database wrapper

Pattern: Multi-step workflow

Pattern: Webhook receiver

Next steps