> ## Documentation Index
> Fetch the complete documentation index at: https://helpcenter.magure.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

<Steps>
  <Step title="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.
  </Step>

  <Step title="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.
  </Step>

  <Step title="Register the MCP server in MagOneAI">
    An **organization owner** connects the server through the [External MCP servers](/tools/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.
  </Step>

  <Step title="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.
  </Step>

  <Step title="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
  </Step>

  <Step title="Attach the tool to agents">
    Enable the custom tool in agent configurations. Agents can now discover and use your custom tools.
  </Step>
</Steps>

## MCP server development basics

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

### Architecture

```mermaid theme={null}
graph LR
    A[Agent] --> B[MagOneAI Platform]
    B --> C[MCP Server]
    C --> D[Your API]
    C --> E[Database]
    C --> F[Internal System]
```

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:

<CardGroup cols={2}>
  <Card title="Python SDK" icon="python">
    ```bash theme={null}
    pip install mcp
    ```

    Best for: Data processing, ML models, database integrations
  </Card>

  <Card title="TypeScript SDK" icon="node-js">
    ```bash theme={null}
    npm install @modelcontextprotocol/sdk
    ```

    Best for: Web APIs, real-time services, Node.js ecosystems
  </Card>
</CardGroup>

### Basic MCP server structure (Python)

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

```python theme={null}
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:

```typescript theme={null}
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

<AccordionGroup>
  <Accordion title="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_data`

    Agents use tool names to understand what the tool does. Clear names improve agent decision-making.
  </Accordion>

  <Accordion title="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"
  </Accordion>

  <Accordion title="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.
  </Accordion>

  <Accordion title="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:**

    ```json theme={null}
    {
      "customer_id": "12345",
      "name": "Acme Corp",
      "status": "active",
      "contacts": [...]
    }
    ```

    ❌ **Bad:**

    ```json theme={null}
    {
      "result": "Customer Acme Corp (ID: 12345) is active"
    }
    ```
  </Accordion>

  <Accordion title="Handle errors gracefully">
    Return structured error responses that agents can interpret:

    ```python theme={null}
    {
        "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).
  </Accordion>
</AccordionGroup>

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

```python theme={null}
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 theme={null}
# 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"]
```

```bash theme={null}
# 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](/tools/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:

```python theme={null}
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:

<Steps>
  <Step title="Configure OAuth app">
    Register an OAuth application with your service provider. Note the client ID and client secret.
  </Step>

  <Step title="Add OAuth config to MagOneAI">
    In Admin Portal, add OAuth configuration:

    * Client ID
    * Client secret
    * Authorization URL
    * Token URL
    * Required scopes
  </Step>

  <Step title="Users authorize access">
    When users first use your tool, they complete the OAuth flow. MagOneAI stores their access and refresh tokens in Vault.
  </Step>

  <Step title="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.
  </Step>
</Steps>

### 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:

```bash theme={null}
# 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

<Tip>
  **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.
</Tip>

## Common patterns and examples

### Pattern: Database wrapper

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

```python theme={null}
@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:

```python theme={null}
@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:

```python theme={null}
# 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

<CardGroup cols={2}>
  <Card title="MCP protocol" icon="link" href="/tools/mcp-protocol">
    Deep dive into how MCP works
  </Card>

  <Card title="Tools overview" icon="plug" href="/tools/overview">
    Learn about MagOneAI's tool architecture
  </Card>

  <Card title="Security best practices" icon="shield" href="/security/overview">
    Secure your custom integrations
  </Card>

  <Card title="Workflow builder" icon="diagram-project" href="/workflows/overview">
    Use custom tools in workflows
  </Card>
</CardGroup>