MagOneAI
DocsChangelogRoadmap
DashboardSupport
Tools & Integrations

Custom MCP tools

Add your own tools and integrations to MagOneAI using the Model Context Protocol

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.

Register the MCP server in MagOneAI

An organization owner connects the server through the External MCP servers flow:

  • The server's HTTPS URL (public endpoints only — private IPs and localhost are rejected)
  • The credential-to-header mapping and credential values the server expects (stored in Vault)

Once connected, the server's tools are discovered automatically and become available to agents across the organization.

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

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:

  1. search_contacts — find contacts by name, email, or company
  2. get_deal_status — retrieve status of a sales opportunity
  3. update_opportunity — modify deal amount, stage, or close date
  4. 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 connect this server in MagOneAI through the External MCP servers flow, using its public HTTPS URL (a plain http:// or private address won't be accepted).

Authentication for custom integrations

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

API key authentication

Keep two credential concerns separate:

  • Your MCP server's own upstream secrets (for example, the CRM API key it uses to call the CRM). These belong to your server's deployment — its environment variables, secret manager, and so on. MagOneAI does not host your server, so it does not manage or inject these.
  • The credential MagOneAI presents to your MCP server. When you connect the server, you map each credential value to the HTTP header the server expects. MagOneAI stores those values in HashiCorp Vault and injects them as request headers on every call to your server — never in logs or exports.

Your MCP server reads its own upstream secrets from its environment as usual:

API_KEY = os.getenv("CRM_API_KEY")  # part of your server's own deployment

OAuth2 for custom services

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

Configure OAuth app

Register an OAuth application with your service provider. Note the client ID and client secret.

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 a request header on the call to your MCP server.

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:

  1. Go to Project Settings → Integrations → Custom Tools → Your Tool
  2. Click Test Tool
  3. Select a tool to test
  4. Enter test parameters
  5. 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

Database tools

Query uploaded CSV and Excel files with SQL automatically, or connect a live PostgreSQL, MySQL, or MongoDB database read-only

Model configuration

Configure LLM providers and models across the MagOneAI platform

MagOneAI© 2026 Magure, Inc.

On this page

OverviewHow to add custom MCP toolsMCP server development basicsArchitectureAvailable SDKsBasic MCP server structure (Python)Basic MCP server structure (TypeScript)Tool definition best practicesExample: Custom CRM integrationTools to implementComplete Python implementationDeploying the MCP serverAuthentication for custom integrationsAPI key authenticationOAuth2 for custom servicesCustom authentication schemesTesting custom toolsLocal testingIntegration testing in MagOneAITesting in workflowsCommon patterns and examplesPattern: Database wrapperPattern: Multi-step workflowPattern: Webhook receiverNext steps