Custom MCP Server Development for AI Agent Integration with Client System

Custom MCP Server Development

Кастомный MCP-сервер предоставляет AI-клиентам (Claude, Cursor, агентам) доступ к корпоративным системам через стандартный протокол. Один MCP-сервер — один раз реализовал, работает со всеми MCP-совместимыми клиентами. MCP SDK доступен для Python, TypeScript, Kotlin, Java, C#.

MCP-сервер на Python

# pip install mcp
from mcp.server.fastmcp import FastMCP
from mcp import types
import json

# Создание сервера
mcp = FastMCP("Corporate CRM Server")

# Tool: вызываемая функция
@mcp.tool()
async def search_customers(
    query: str,
    limit: int = 10,
    status: str = "active",
) -> list[dict]:
    """Поиск клиентов в CRM по имени или компании.

    Args:
        query: Поисковый запрос (имя или название компании)
        limit: Максимальное количество результатов (1-50)
        status: Статус клиентов (active, inactive, all)
    """
    results = await crm_db.search(
        query=query,
        limit=min(limit, 50),
        status=status,
    )
    return [{"id": r.id, "name": r.name, "company": r.company, "email": r.email}
            for r in results]

@mcp.tool()
async def get_customer_orders(customer_id: str, months: int = 3) -> dict:
    """Получить историю заказов клиента за последние N месяцев.

    Args:
        customer_id: ID клиента в CRM
        months: Количество месяцев истории (1-24)
    """
    orders = await orders_db.get_for_customer(customer_id, months=months)
    return {
        "customer_id": customer_id,
        "total_orders": len(orders),
        "total_revenue": sum(o.amount for o in orders),
        "orders": [{"id": o.id, "date": str(o.date), "amount": o.amount, "status": o.status}
                   for o in orders[:20]],
    }

@mcp.tool()
async def create_task(
    customer_id: str,
    title: str,
    description: str,
    assignee: str,
    due_date: str,
) -> dict:
    """Создать задачу в CRM для менеджера.

    Args:
        customer_id: ID клиента
        title: Заголовок задачи
        description: Подробное описание
        assignee: Email менеджера-исполнителя
        due_date: Дата выполнения в формате YYYY-MM-DD
    """
    task = await crm_tasks.create(
        customer_id=customer_id,
        title=title,
        description=description,
        assignee=assignee,
        due_date=due_date,
    )
    return {"task_id": task.id, "url": task.url}

# Resource: читаемые данные
@mcp.resource("crm://dashboards/{dashboard_id}")
async def get_dashboard(dashboard_id: str) -> str:
    """Дашборд CRM с ключевыми метриками"""
    data = await crm_analytics.get_dashboard(dashboard_id)
    return json.dumps(data, ensure_ascii=False, indent=2)

@mcp.resource("crm://segments")
async def list_segments() -> str:
    """Список клиентских сегментов"""
    segments = await crm_db.get_segments()
    return json.dumps([{"id": s.id, "name": s.name, "count": s.count}
                       for s in segments], ensure_ascii=False)

# Prompt: шаблоны для AI
@mcp.prompt()
def customer_analysis_prompt(customer_id: str) -> list[types.PromptMessage]:
    """Шаблон для глубокого анализа клиента"""
    return [
        types.PromptMessage(
            role="user",
            content=types.TextContent(
                type="text",
                text=f"""Проведи полный анализ клиента {customer_id}:
1. История заказов и тренды
2. Риск оттока (на основе паттернов активности)
3. Возможности для upsell
4. Рекомендуемые следующие действия"""
            ),
        )
    ]

# Запуск сервера
if __name__ == "__main__":
    mcp.run()  # stdio транспорт по умолчанию

MCP-сервер с HTTP транспортом

# Для удалённого доступа — HTTP+SSE транспорт
from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI
import uvicorn

mcp = FastMCP("Remote CRM Server")

# ... определение tools, resources, prompts ...

# Создание ASGI-приложения
app = mcp.get_asgi_app()

# Добавляем middleware для авторизации
from fastapi.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not await verify_token(token):
            return Response(status_code=401)
        return await call_next(request)

app.add_middleware(AuthMiddleware)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

TypeScript MCP-сервер

import { McpServer, ResourceTemplate } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "corporate-analytics",
  version: "1.0.0",
});

// Tool с типизацией через Zod
server.tool(
  "query_analytics",
  "Запрос к аналитической базе данных",
  {
    sql: z.string().describe("SQL SELECT-запрос"),
    limit: z.number().max(1000).default(100),
  },
  async ({ sql, limit }) => {
    const result = await analyticsDb.query(`${sql} LIMIT ${limit}`);
    return {
      content: [{
        type: "text",
        text: JSON.stringify(result.rows),
      }],
    };
  }
);

// Resource с динамическим URI
server.resource(
  "report",
  new ResourceTemplate("analytics://reports/{report_id}", { list: undefined }),
  async (uri, { report_id }) => {
    const report = await reportService.get(report_id);
    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(report),
      }],
    };
  }
);

// Запуск
const transport = new StdioServerTransport();
await server.connect(transport);

Безопасность и разграничение доступа

from functools import wraps
from mcp.server.fastmcp import FastMCP, Context

mcp = FastMCP("Secure CRM Server")

def require_permission(permission: str):
    """Декоратор для проверки прав доступа"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, ctx: Context, **kwargs):
            user_permissions = ctx.request_context.meta.get("permissions", [])
            if permission not in user_permissions:
                raise PermissionError(f"Требуется право: {permission}")
            return await func(*args, ctx=ctx, **kwargs)
        return wrapper
    return decorator

@mcp.tool()
@require_permission("crm:write")
async def update_customer(customer_id: str, data: dict, ctx: Context) -> dict:
    """Обновить данные клиента (требует право crm:write)"""
    return await crm_db.update(customer_id, data)

Практический кейс: CRM-интеграция для отдела продаж

Задача: менеджеры по продажам использовали Claude для подготовки к звонкам, но не имели быстрого доступа к данным клиентов — нужно было переключаться в CRM.

MCP-сервер: подключён к Bitrix24 через REST API, предоставляет инструменты: поиск клиентов, история сделок, задачи, аналитика по сегментам.

Настройка: один MCP-сервер на корпоративном сервере → все сотрудники подключаются через Claude Desktop с авторизацией.

Результаты:

• Время подготовки к звонку: 15 мин → 4 мин
• Переключений между Claude и CRM: -83%
• Внедрение без дополнительного обучения (Claude понимает естественные вопросы)

Сроки

• Базовый MCP-сервер (stdio): 1–2 дня
• HTTP-транспорт с авторизацией: 2–3 дня
• Полная интеграция с корпоративной системой: 1–2 недели
• Развёртывание и настройка для команды: 2–3 дня