Как написать свой MCP сервер на Python: пошаговый туториал

Готовых MCP серверов — тысячи. GitHub, Slack, PostgreSQL, Jira, Notion — для популярных сервисов серверы уже написаны и работают. Но рано или поздно вы упираетесь в ситуацию, когда готового сервера нет. Внутренний API компании. Кастомная CRM. Самописная система деплоя. Специфичная бизнес-логика, которую никто за вас не обернёт.

В такой момент нужно писать свой MCP сервер. Это проще, чем кажется: минимальный сервер на Python — 30 строк кода. Полноценный production-сервер с аутентификацией, error handling и тестами — 200-300 строк.

Эта статья — пошаговый туториал. Напишем MCP сервер для работы с Jira: от первого @mcp.tool() до публикации pip-пакета. Весь код на Python с использованием FastMCP — высокоуровневого API из официального SDK.

Зачем свой MCP сервер

Готовых серверов не хватает в нескольких типичных ситуациях:

Внутренние API. У компании своя система тикетов, свой CI/CD, свой дашборд метрик. Готового MCP сервера для них не будет — нужно писать обёртку самостоятельно.

Специфичная бизнес-логика. Готовый MCP сервер для Jira существует. Но вам нужен не просто доступ к тикетам — нужна логика: автоматическое назначение спринта по приоритету, расчёт velocity команды, генерация release notes из закрытых тикетов. Это не универсальный сервер, а ваш инструмент под ваш процесс.

Композитные инструменты. Один MCP сервер может объединять несколько источников данных. Сервер для onboarding-процесса: тянет данные из HR-системы, создаёт задачи в Jira, настраивает доступы через внутренний API, отправляет приветственное сообщение в Slack. Один tool — один промпт — одно действие AI — похожая оркестрация встречается и в мультиагентных архитектурах, где несколько специализированных агентов делят общий набор инструментов.

Что построим

MCP сервер для работы с Jira. Четыре инструмента:

get_issue — получить данные тикета по ключу
search_issues — поиск тикетов по JQL
create_issue — создать новый тикет
add_comment — добавить комментарий к тикету

Плюс resources для чтения данных и prompts для шаблонов взаимодействия. Полноценный сервер, который подключается к Claude Code и даёт AI работать с вашим Jira-проектом через естественный язык.

Стек: Python 3.10+, mcp SDK (FastMCP), httpx для HTTP-запросов.

Шаг 1: Установка и структура проекта

Установка SDK

pip install "mcp>=1.28,<2" "fastmcp>=2.0" httpx

Или через uv (рекомендуется для новых проектов):

uv init mcp-jira-server
cd mcp-jira-server
uv add "mcp>=1.28,<2" "fastmcp>=2.0" httpx

Два пакета работают в связке: mcp — официальный Python SDK от Anthropic (v1.28+, стабильный; v2 в alpha — пока не использовать в production, добавляйте ограничение <2) с базовым протоколом; fastmcp — самостоятельный пакет (v2.x, от jlowin/fastmcp) с высокоуровневым FastMCP API, клиентом, утилитами тестирования и расширенным API промптов. FastMCP — рекомендуемый способ писать MCP серверы: он включает официальный mcp как зависимость.

Структура проекта

mcp-jira-server/
├── src/
│   └── mcp_jira/
│       ├── __init__.py
│       ├── server.py          # FastMCP сервер
│       ├── jira_client.py     # HTTP-клиент для Jira API
│       └── config.py          # Конфигурация
├── tests/
│   ├── __init__.py
│   └── test_server.py
├── pyproject.toml
└── README.md

Минимальный сервер

Начнём с самого простого — убедимся, что SDK работает:

# src/mcp_jira/server.py
from fastmcp import FastMCP

mcp = FastMCP("Jira MCP Server")


@mcp.tool
def ping() -> str:
    """Проверка работоспособности сервера"""
    return "pong"


if __name__ == "__main__":
    mcp.run()

Запуск:

python src/mcp_jira/server.py

Сервер запускается с транспортом stdio — стандартный способ для локальных MCP серверов. Ввод/вывод через stdin/stdout, протокол JSON-RPC 2.0. Для удалённых серверов (API-сервис с несколькими клиентами) используется Streamable HTTP транспорт: mcp.run(transport="http", host="0.0.0.0", port=8000). В FastMCP 3 параметры транспорта (host, port, log_level и другие) передаются в run(), а не в конструктор FastMCP().

Шаг 2: Первый tool — работа с тикетами

Tools — основная единица MCP сервера. Это функции, которые AI-модель может вызвать для выполнения действий с побочными эффектами: создание, обновление, удаление. Resources, напротив, предназначены только для чтения данных.

Jira HTTP-клиент

Сначала обёртка над Jira REST API:

# src/mcp_jira/jira_client.py
import httpx
from dataclasses import dataclass


@dataclass
class JiraConfig:
    base_url: str       # https://your-company.atlassian.net
    email: str          # [email protected]
    api_token: str      # Jira API token


class JiraClient:
    def __init__(self, config: JiraConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=f"{config.base_url}/rest/api/3",
            auth=(config.email, config.api_token),
            headers={"Content-Type": "application/json"},
            timeout=30.0,
        )

    async def get_issue(self, issue_key: str) -> dict:
        """Получить тикет по ключу (например, PROJ-123)"""
        response = await self.client.get(f"/issue/{issue_key}")
        response.raise_for_status()
        return response.json()

    async def search_issues(self, jql: str, max_results: int = 20) -> list[dict]:
        """Поиск тикетов по JQL"""
        response = await self.client.post(
            "/search",
            json={
                "jql": jql,
                "maxResults": max_results,
                "fields": [
                    "summary", "status", "assignee",
                    "priority", "created", "updated",
                ],
            },
        )
        response.raise_for_status()
        data = response.json()
        return data.get("issues", [])

    async def create_issue(
        self,
        project_key: str,
        summary: str,
        description: str,
        issue_type: str = "Task",
    ) -> dict:
        """Создать новый тикет"""
        response = await self.client.post(
            "/issue",
            json={
                "fields": {
                    "project": {"key": project_key},
                    "summary": summary,
                    "description": {
                        "type": "doc",
                        "version": 1,
                        "content": [
                            {
                                "type": "paragraph",
                                "content": [
                                    {"type": "text", "text": description}
                                ],
                            }
                        ],
                    },
                    "issuetype": {"name": issue_type},
                },
            },
        )
        response.raise_for_status()
        return response.json()

    async def add_comment(self, issue_key: str, body: str) -> dict:
        """Добавить комментарий к тикету"""
        response = await self.client.post(
            f"/issue/{issue_key}/comment",
            json={
                "body": {
                    "type": "doc",
                    "version": 1,
                    "content": [
                        {
                            "type": "paragraph",
                            "content": [
                                {"type": "text", "text": body}
                            ],
                        }
                    ],
                },
            },
        )
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self.client.aclose()

Регистрация tools

Подключаем инструменты к MCP серверу. Каждый tool — функция с декоратором @mcp.tool(). FastMCP автоматически генерирует JSON Schema из аннотаций типов и docstring.

# src/mcp_jira/server.py
import os
from fastmcp import FastMCP
from mcp_jira.jira_client import JiraClient, JiraConfig

mcp = FastMCP("Jira MCP Server")

# Инициализация клиента из переменных окружения
jira = JiraClient(
    JiraConfig(
        base_url=os.environ["JIRA_BASE_URL"],
        email=os.environ["JIRA_EMAIL"],
        api_token=os.environ["JIRA_API_TOKEN"],
    )
)


@mcp.tool()
async def get_issue(issue_key: str) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    data = await jira.get_issue(issue_key)
    fields = data["fields"]
    assignee = fields.get("assignee")
    assignee_name = assignee.get("displayName", "Не назначен") if assignee else "Не назначен"
    return (
        f"**{data['key']}**: {fields['summary']}\n"
        f"Статус: {fields['status']['name']}\n"
        f"Приоритет: {fields.get('priority', {}).get('name', 'Не указан')}\n"
        f"Назначен: {assignee_name}\n"
        f"Создан: {fields['created']}\n"
        f"Обновлён: {fields['updated']}"
    )


@mcp.tool()
async def search_issues(jql: str, max_results: int = 20) -> str:
    """Поиск тикетов Jira по JQL-запросу.

    Args:
        jql: JQL-запрос, например 'project = PROJ AND status = "In Progress"'
        max_results: Максимальное количество результатов (по умолчанию 20)
    """
    issues = await jira.search_issues(jql, max_results)
    if not issues:
        return "Тикеты не найдены."

    lines = []
    for issue in issues:
        fields = issue["fields"]
        assignee = fields.get("assignee", {})
        assignee_name = assignee.get("displayName", "—") if assignee else "—"
        lines.append(
            f"- **{issue['key']}**: {fields['summary']} "
            f"[{fields['status']['name']}] → {assignee_name}"
        )
    return f"Найдено {len(issues)} тикетов:\n\n" + "\n".join(lines)


@mcp.tool()
async def create_issue(
    project_key: str,
    summary: str,
    description: str = "",
    issue_type: str = "Task",
) -> str:
    """Создать новый тикет в Jira.

    Args:
        project_key: Ключ проекта, например PROJ
        summary: Заголовок тикета
        description: Описание тикета
        issue_type: Тип тикета — Task, Bug, Story (по умолчанию Task)
    """
    result = await jira.create_issue(
        project_key, summary, description, issue_type
    )
    return f"Создан тикет **{result['key']}**: {summary}"


@mcp.tool()
async def add_comment(issue_key: str, body: str) -> str:
    """Добавить комментарий к тикету Jira.

    Args:
        issue_key: Ключ тикета, например PROJ-123
        body: Текст комментария
    """
    await jira.add_comment(issue_key, body)
    return f"Комментарий добавлен к {issue_key}."


if __name__ == "__main__":
    mcp.run()

Ключевые моменты:

Docstring = описание для AI. FastMCP берёт docstring функции и передаёт AI-модели как описание инструмента. Модель решает, какой tool вызвать, на основе этого описания — тот же принцип лежит в основе хороших промптов в целом, разобранный в нашем гайде по написанию эффективных промптов. Пишите docstring так, чтобы AI понимал назначение и параметры.

Аннотации типов = JSON Schema. Типы параметров (str, int, значения по умолчанию) автоматически конвертируются в JSON Schema. AI получает типизированную схему ввода — знает, что передавать.

Возвращаемое значение — строка. Tools возвращают текст, который AI-модель получает как результат вызова. Форматируйте для читаемости: markdown работает.

Шаг 3: Resources — данные для чтения

Resources — второй тип capability в MCP. Семантически это GET-запросы: только чтение, без побочных эффектов. AI-модель читает resources для получения контекста перед тем, как принять решение.

@mcp.resource("jira://projects")
async def list_projects() -> str:
    """Список доступных проектов Jira"""
    response = await jira.client.get("/project")
    response.raise_for_status()
    projects = response.json()

    lines = []
    for proj in projects:
        lines.append(f"- **{proj['key']}**: {proj['name']}")
    return "\n".join(lines)

URI templates — динамические resources

Статические resources полезны для фиксированных данных. Для параметризованных данных в MCP есть URI templates:

@mcp.resource("jira://issue/{issue_key}")
async def get_issue_resource(issue_key: str) -> str:
    """Данные тикета Jira для контекста.

    Используется AI для получения информации о тикете
    без выполнения действий.
    """
    data = await jira.get_issue(issue_key)
    fields = data["fields"]

    return (
        f"# {data['key']}: {fields['summary']}\n\n"
        f"**Статус:** {fields['status']['name']}\n"
        f"**Приоритет:** {fields.get('priority', {}).get('name', 'N/A')}\n"
        f"**Тип:** {fields['issuetype']['name']}\n"
        f"**Назначен:** {fields.get('assignee', {}).get('displayName', 'Не назначен')}\n\n"
        f"**Описание:**\n{_extract_text(fields.get('description', {}))}"
    )


def _extract_text(adf: dict) -> str:
    """Извлечь текст из Atlassian Document Format"""
    if not adf or not isinstance(adf, dict):
        return "Нет описания"

    texts = []
    for block in adf.get("content", []):
        for inline in block.get("content", []):
            if inline.get("type") == "text":
                texts.append(inline["text"])
    return "\n".join(texts) if texts else "Нет описания"


@mcp.resource("jira://sprint/{board_id}/active")
async def get_active_sprint(board_id: str) -> str:
    """Данные активного спринта для доски.

    Args:
        board_id: ID доски в Jira (Agile board)
    """
    # Sprint endpoint относится к Jira Agile API (/rest/agile/1.0/),
    # а не к /rest/api/3 — поэтому строим полный URL
    response = await jira.client.get(
        f"{jira.config.base_url}/rest/agile/1.0/board/{board_id}/sprint",
        params={"state": "active"},
    )
    response.raise_for_status()
    sprints = response.json().get("values", [])

    if not sprints:
        return "Нет активного спринта."

    sprint = sprints[0]
    return (
        f"# Спринт: {sprint['name']}\n\n"
        f"**Цель:** {sprint.get('goal', 'Не указана')}\n"
        f"**Начало:** {sprint.get('startDate', 'N/A')}\n"
        f"**Окончание:** {sprint.get('endDate', 'N/A')}\n"
        f"**Статус:** {sprint['state']}"
    )

Когда tool, когда resource

Различие принципиальное:

	Tool	Resource
Назначение	Выполнить действие	Предоставить данные для чтения
Побочные эффекты	Да (создание, обновление)	Нет
Аналогия REST	POST, PUT, DELETE	GET
Пример	Создать тикет, добавить комментарий	Данные тикета, список проектов

На практике: get_issue как tool и как resource делают похожие вещи. Разница в семантике. Tool — для явного вызова в рамках действия. Resource — для получения контекста перед принятием решения.

Шаг 4: Prompts — шаблоны взаимодействия

Prompts — третий тип capability. Это шаблоны промптов, которые MCP сервер предлагает AI-модели. Клиент (Claude Code, Cursor) использует их как стартовую точку для взаимодействия.

from fastmcp.prompts import Message


@mcp.prompt()
def triage_issue(issue_key: str) -> str:
    """Проанализировать тикет и предложить приоритет и назначение.

    Args:
        issue_key: Ключ тикета для анализа
    """
    return (
        f"Проанализируй тикет {issue_key} в Jira.\n\n"
        f"1. Прочитай данные тикета через get_issue\n"
        f"2. Определи приоритет на основе описания и типа\n"
        f"3. Предложи, кому назначить, исходя из текущей загрузки команды\n"
        f"4. Оцени трудоёмкость в story points\n\n"
        f"Верни структурированный ответ с обоснованием каждого решения."
    )


@mcp.prompt()
def sprint_review() -> str:
    """Сформировать отчёт по текущему спринту"""
    return (
        "Сформируй отчёт по текущему спринту:\n\n"
        "1. Найди все тикеты текущего спринта через search_issues\n"
        "2. Сгруппируй по статусам: Done, In Progress, To Do\n"
        "3. Посчитай процент выполнения\n"
        "4. Выдели заблокированные тикеты\n"
        "5. Сформулируй риски для завершения спринта\n\n"
        "Формат: markdown-таблица со статистикой + текстовый анализ рисков."
    )


@mcp.prompt()
def release_notes(version: str) -> list[Message]:
    """Сгенерировать release notes для версии.

    Args:
        version: Номер версии, например 2.4.0
    """
    return [
        Message(
            f"Сгенерируй release notes для версии {version}.\n\n"
            f"Найди все тикеты с fixVersion = {version} через search_issues "
            f'с JQL: fixVersion = "{version}" AND status = Done\n\n'
            f"Сгруппируй по типам: Features, Bug Fixes, Improvements.\n"
            f"Формат: markdown, готовый для публикации."
        ),
        Message(
            f"Начну формировать release notes для v{version}. "
            f"Сначала найду все закрытые тикеты этой версии.",
            role="assistant",
        ),
    ]

Prompts возвращают либо строку (однозвеньевой промпт), либо список Message (мультитёрн — диалог с предзаданными репликами). Второй вариант полезен, когда нужно задать контекст ассистенту перед выполнением задачи.

Шаг 5: Подключение к Claude Code

Сервер написан. Подключаем его к Claude Code.

Конфигурация через CLI

Самый простой способ — команда claude mcp add:

claude mcp add jira-server \
  -e JIRA_BASE_URL=https://your-company.atlassian.net \
  -e [email protected] \
  -e JIRA_API_TOKEN=your-api-token \
  -- python src/mcp_jira/server.py

Claude Code добавит запись в файл конфигурации и перезапустит сервер.

Конфигурация через JSON

Альтернатива — файл .mcp.json в корне проекта:

{
  "mcpServers": {
    "jira-server": {
      "command": "python",
      "args": ["src/mcp_jira/server.py"],
      "env": {
        "JIRA_BASE_URL": "https://your-company.atlassian.net",
        "JIRA_EMAIL": "[email protected]",
        "JIRA_API_TOKEN": "your-api-token"
      }
    }
  }
}

Если используете uv:

{
  "mcpServers": {
    "jira-server": {
      "command": "uv",
      "args": ["run", "python", "src/mcp_jira/server.py"],
      "env": {
        "JIRA_BASE_URL": "https://your-company.atlassian.net",
        "JIRA_EMAIL": "[email protected]",
        "JIRA_API_TOKEN": "your-api-token"
      }
    }
  }
}

Проверка подключения

После перезапуска Claude Code сервер появляется в списке доступных tools. Проверьте:

> claude mcp list

Должен отобразиться jira-server со статусом connected и списком зарегистрированных tools.

Теперь можно использовать в диалоге:

> Покажи все открытые баги в проекте BACKEND

Claude Code увидит tool search_issues, сформирует JQL project = BACKEND AND type = Bug AND status != Done, вызовет инструмент и отобразит результат.

Шаг 6: Error handling

Production-сервер не должен падать при первой ошибке. Выстраиваем защиту послойно.

Уровень 1: Валидация входных данных

import re


@mcp.tool()
async def get_issue(issue_key: str) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    # Валидация формата ключа
    if not re.match(r"^[A-Z][A-Z0-9_]+-\d+$", issue_key):
        return f"Некорректный формат ключа: {issue_key}. Ожидается формат PROJ-123."

    data = await jira.get_issue(issue_key)
    # ... форматирование

Возвращаем текст ошибки, а не бросаем исключение. AI-модель получает сообщение и корректирует запрос.

Уровень 2: Обработка HTTP-ошибок

import httpx


@mcp.tool()
async def get_issue(issue_key: str) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    if not re.match(r"^[A-Z][A-Z0-9_]+-\d+$", issue_key):
        return f"Некорректный формат ключа: {issue_key}. Ожидается формат PROJ-123."

    try:
        data = await jira.get_issue(issue_key)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return f"Тикет {issue_key} не найден."
        if e.response.status_code == 401:
            return "Ошибка аутентификации. Проверьте JIRA_API_TOKEN."
        if e.response.status_code == 403:
            return f"Нет доступа к тикету {issue_key}."
        return f"Ошибка Jira API: {e.response.status_code}"
    except httpx.ConnectError:
        return "Не удалось подключиться к Jira. Проверьте JIRA_BASE_URL."
    except httpx.TimeoutException:
        return "Таймаут при запросе к Jira. Попробуйте позже."

    fields = data["fields"]
    return (
        f"**{data['key']}**: {fields['summary']}\n"
        f"Статус: {fields['status']['name']}\n"
        f"Приоритет: {fields.get('priority', {}).get('name', 'Не указан')}\n"
        f"Назначен: {fields.get('assignee', {}).get('displayName', 'Не назначен')}\n"
        f"Создан: {fields['created']}\n"
        f"Обновлён: {fields['updated']}"
    )

Уровень 3: Общий обработчик через декоратор

Чтобы не дублировать try/except в каждом tool, вынесите обработку в декоратор:

import functools
import logging

logger = logging.getLogger("mcp_jira")


def handle_errors(func):
    """Декоратор для обработки ошибок в MCP tools"""
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            error_map = {
                401: "Ошибка аутентификации. Проверьте JIRA_API_TOKEN.",
                403: "Нет доступа. Проверьте права учётной записи.",
                404: "Ресурс не найден.",
                429: "Превышен лимит запросов. Подождите и повторите.",
            }
            msg = error_map.get(status, f"Ошибка Jira API: {status}")
            logger.error(f"HTTP {status} in {func.__name__}: {e}")
            return msg
        except httpx.ConnectError:
            logger.error(f"Connection error in {func.__name__}")
            return "Не удалось подключиться к Jira. Проверьте JIRA_BASE_URL."
        except httpx.TimeoutException:
            logger.error(f"Timeout in {func.__name__}")
            return "Таймаут при запросе к Jira."
        except Exception as e:
            logger.exception(f"Unexpected error in {func.__name__}")
            return f"Внутренняя ошибка: {type(e).__name__}"
    return wrapper


@mcp.tool()
@handle_errors
async def get_issue(issue_key: str) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    data = await jira.get_issue(issue_key)
    fields = data["fields"]
    return (
        f"**{data['key']}**: {fields['summary']}\n"
        f"Статус: {fields['status']['name']}\n"
        f"Приоритет: {fields.get('priority', {}).get('name', 'Не указан')}\n"
        f"Назначен: {fields.get('assignee', {}).get('displayName', 'Не назначен')}"
    )

Принцип: tool никогда не бросает необработанное исключение. AI-модель всегда получает текстовый ответ — либо данные, либо понятное описание ошибки. После запуска сервера та же дисциплина нужна и в наблюдаемости: трассировка каждого вызова tool через что-то вроде Langfuse делает молчаливые сбои видимыми, а не погребёнными в логах.

Шаг 7: Аутентификация и secrets

В примерах выше мы читали credentials из переменных окружения. Это базовый подход. Для production есть несколько вариантов.

Вариант 1: Переменные окружения (базовый)

# src/mcp_jira/config.py
import os


def get_jira_config() -> JiraConfig:
    base_url = os.environ.get("JIRA_BASE_URL")
    email = os.environ.get("JIRA_EMAIL")
    api_token = os.environ.get("JIRA_API_TOKEN")

    if not all([base_url, email, api_token]):
        missing = []
        if not base_url:
            missing.append("JIRA_BASE_URL")
        if not email:
            missing.append("JIRA_EMAIL")
        if not api_token:
            missing.append("JIRA_API_TOKEN")
        raise ValueError(
            f"Не заданы переменные окружения: {', '.join(missing)}"
        )

    return JiraConfig(
        base_url=base_url,
        email=email,
        api_token=api_token,
    )

В .mcp.json передаёте через секцию env. Не коммитьте реальные токены — используйте .env.local или менеджер секретов.

Вариант 2: Lifespan с инициализацией

FastMCP поддерживает lifespan — паттерн для инициализации/очистки ресурсов при старте/остановке сервера. Идеально для клиентов с подключениями:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from fastmcp import FastMCP, Context


@dataclass
class AppContext:
    jira: JiraClient


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    """Инициализация при старте, очистка при остановке"""
    config = get_jira_config()
    jira = JiraClient(config)
    try:
        yield AppContext(jira=jira)
    finally:
        await jira.close()


mcp = FastMCP("Jira MCP Server", lifespan=app_lifespan)


@mcp.tool()
@handle_errors
async def get_issue(issue_key: str, ctx: Context) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    jira = ctx.lifespan_context.jira
    data = await jira.get_issue(issue_key)
    fields = data["fields"]
    return (
        f"**{data['key']}**: {fields['summary']}\n"
        f"Статус: {fields['status']['name']}"
    )

Context — объект, который FastMCP передаёт в tool-функцию. Через ctx.lifespan_context доступен объект, возвращённый lifespan-функцией (AppContext с инициализированными зависимостями). Параметр ctx: Context автоматически инжектится фреймворком — AI-модель его не видит и не передаёт.

Вариант 3: OAuth 2.1 (для удалённых серверов)

Для серверов, доступных по HTTP (не stdio), MCP SDK поддерживает OAuth 2.1:

from pydantic import AnyHttpUrl
from fastmcp.server.auth import AccessToken, TokenVerifier, AuthSettings


class JiraTokenVerifier(TokenVerifier):
    async def verify_token(self, token: str) -> AccessToken | None:
        # Валидация токена через вашу систему аутентификации
        # Возвращаете AccessToken при успехе, None при ошибке
        pass


mcp = FastMCP(
    "Jira MCP Server",
    token_verifier=JiraTokenVerifier(),
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.your-company.com"),
        resource_server_url=AnyHttpUrl("https://mcp-jira.your-company.com"),
        required_scopes=["jira:read", "jira:write"],
    ),
)

OAuth нужен, когда MCP сервер развёрнут как удалённый сервис (Streamable HTTP транспорт) и к нему подключаются несколько пользователей. Для локальных серверов (stdio) переменных окружения достаточно.

Шаг 8: Тестирование

MCP Inspector

MCP Inspector — интерактивный UI для тестирования серверов. Подключается к серверу, показывает список tools, resources, prompts, позволяет вызывать их вручную.

npx -y @modelcontextprotocol/inspector

Inspector откроет браузер с интерфейсом. Укажите команду запуска сервера — Inspector подключится через stdio и покажет все зарегистрированные capabilities.

Что проверять:

Все tools отображаются с корректными описаниями
JSON Schema параметров соответствует ожидаемому
Вызов каждого tool возвращает корректный результат
Ошибки возвращают понятные сообщения, а не трейсбеки

Unit-тесты с pytest

MCP SDK предоставляет класс Client для тестирования серверов без запуска транспорта:

# tests/test_server.py
import pytest
from unittest.mock import AsyncMock, patch
from fastmcp.client import Client

from mcp_jira.server import mcp
from mcp_jira.jira_client import JiraClient


@pytest.fixture
async def client():
    async with Client(mcp) as c:
        yield c


async def test_get_issue(client: Client):
    """Тест получения тикета"""
    mock_response = {
        "key": "PROJ-123",
        "fields": {
            "summary": "Fix login bug",
            "status": {"name": "In Progress"},
            "priority": {"name": "High"},
            "assignee": {"displayName": "John Doe"},
            "issuetype": {"name": "Bug"},
            "description": {},
            "created": "2026-01-15T10:00:00.000+0000",
            "updated": "2026-01-16T14:30:00.000+0000",
        },
    }

    # Патчим метод класса — работает с lifespan-архитектурой,
    # где JiraClient инициализируется внутри AppContext
    with patch.object(JiraClient, "get_issue", new_callable=AsyncMock, return_value=mock_response):
        result = await client.call_tool("get_issue", {"issue_key": "PROJ-123"})
        text = result.content[0].text
        assert "PROJ-123" in text
        assert "Fix login bug" in text
        assert "In Progress" in text


async def test_search_issues_empty(client: Client):
    """Тест поиска без результатов"""
    with patch.object(JiraClient, "search_issues", new_callable=AsyncMock, return_value=[]):
        result = await client.call_tool(
            "search_issues",
            {"jql": "project = EMPTY"},
        )
        assert "не найдены" in result.content[0].text


async def test_create_issue(client: Client):
    """Тест создания тикета"""
    mock_response = {"key": "PROJ-456", "id": "10456"}

    with patch.object(JiraClient, "create_issue", new_callable=AsyncMock, return_value=mock_response):
        result = await client.call_tool(
            "create_issue",
            {
                "project_key": "PROJ",
                "summary": "New feature request",
                "description": "Add dark mode support",
            },
        )
        assert "PROJ-456" in result.content[0].text


async def test_get_issue_not_found(client: Client):
    """Тест обработки 404"""
    import httpx

    mock_response = httpx.Response(404, request=httpx.Request("GET", "test"))
    error = httpx.HTTPStatusError(
        "Not found", request=mock_response.request, response=mock_response
    )

    with patch.object(JiraClient, "get_issue", new_callable=AsyncMock, side_effect=error):
        result = await client.call_tool(
            "get_issue", {"issue_key": "PROJ-999"}
        )
        assert "не найден" in result.content[0].text

Запуск:

pip install "pytest>=8" pytest-asyncio "fastmcp>=2.0"
pytest tests/ -v

Добавьте в pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"

Client(mcp) из пакета fastmcp подключается к серверу in-process, без сетевого транспорта. result.content[0].text содержит текст ответа tool; result.data — гидратированный Python-объект (доступен, если tool объявляет output_schema). Тесты быстрые (миллисекунды), не требуют запущенного Jira.

Тестирование resources и prompts

import httpx

async def test_list_projects_resource(client: Client):
    """Тест resource со списком проектов"""
    mock_projects = [
        {"key": "PROJ", "name": "Main Project"},
        {"key": "INFRA", "name": "Infrastructure"},
    ]

    # В lifespan-архитектуре глобального jira нет — патчим метод httpx.AsyncClient
    with patch.object(
        httpx.AsyncClient, "get", new_callable=AsyncMock
    ) as mock_get:
        mock_get.return_value.json.return_value = mock_projects
        mock_get.return_value.raise_for_status = lambda: None

        result = await client.read_resource("jira://projects")
        text = result[0].text
        assert "PROJ" in text
        assert "INFRA" in text


async def test_triage_prompt(client: Client):
    """Тест prompt для триажа"""
    result = await client.get_prompt(
        "triage_issue", {"issue_key": "PROJ-123"}
    )
    text = result.messages[0].content.text
    assert "PROJ-123" in text
    assert "приоритет" in text

Шаг 9: Публикация

MCP сервер можно распространять несколькими способами — от pip-пакета до Docker-образа.

Способ 1: pip-пакет

Оформите проект как Python-пакет:

# pyproject.toml
[project]
name = "mcp-jira-server"
version = "0.1.0"
description = "MCP server for Jira integration"
requires-python = ">=3.10"
dependencies = [
    "mcp>=1.28.0,<2",
    "fastmcp>=2.0.0",
    "httpx>=0.28.0",
]

[project.scripts]
mcp-jira = "mcp_jira.server:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

Добавьте entry point в сервер:

# src/mcp_jira/server.py

def main():
    mcp.run()

if __name__ == "__main__":
    main()

Публикация:

pip install build twine
python -m build
twine upload dist/*

После установки (pip install mcp-jira-server) пользователь подключает через:

{
  "mcpServers": {
    "jira": {
      "command": "mcp-jira",
      "env": {
        "JIRA_BASE_URL": "https://company.atlassian.net",
        "JIRA_EMAIL": "[email protected]",
        "JIRA_API_TOKEN": "token"
      }
    }
  }
}

Способ 2: Docker

FROM python:3.12-slim

WORKDIR /app

COPY pyproject.toml .
COPY src/ src/

RUN pip install --no-cache-dir .

CMD ["mcp-jira"]

docker build -t mcp-jira-server .

Конфигурация для Claude Code:

{
  "mcpServers": {
    "jira": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-e", "JIRA_BASE_URL",
        "-e", "JIRA_EMAIL",
        "-e", "JIRA_API_TOKEN",
        "mcp-jira-server"
      ],
      "env": {
        "JIRA_BASE_URL": "https://company.atlassian.net",
        "JIRA_EMAIL": "[email protected]",
        "JIRA_API_TOKEN": "token"
      }
    }
  }
}

Флаг -i критически важен — он подключает stdin к контейнеру, что необходимо для stdio-транспорта.

Способ 3: uvx (zero-install)

Если пакет опубликован на PyPI, пользователи запускают без явной установки:

{
  "mcpServers": {
    "jira": {
      "command": "uvx",
      "args": ["mcp-jira-server"],
      "env": {
        "JIRA_BASE_URL": "https://company.atlassian.net",
        "JIRA_EMAIL": "[email protected]",
        "JIRA_API_TOKEN": "token"
      }
    }
  }
}

uvx скачает пакет, создаст изолированное окружение и запустит сервер. Без pip install, без виртуальных окружений.

Полный код сервера

Финальная версия со всеми компонентами:

# src/mcp_jira/server.py
"""MCP сервер для работы с Jira."""

import functools
import logging
import os
import re
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

import httpx
from fastmcp import Context, FastMCP
from fastmcp.prompts import Message

# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logger = logging.getLogger("mcp_jira")

# ---------------------------------------------------------------------------
# Jira client
# ---------------------------------------------------------------------------

@dataclass
class JiraConfig:
    base_url: str
    email: str
    api_token: str


class JiraClient:
    def __init__(self, config: JiraConfig):
        self.config = config
        self.client = httpx.AsyncClient(
            base_url=f"{config.base_url}/rest/api/3",
            auth=(config.email, config.api_token),
            headers={"Content-Type": "application/json"},
            timeout=30.0,
        )

    async def get_issue(self, issue_key: str) -> dict:
        response = await self.client.get(f"/issue/{issue_key}")
        response.raise_for_status()
        return response.json()

    async def search_issues(self, jql: str, max_results: int = 20) -> list[dict]:
        response = await self.client.post(
            "/search",
            json={
                "jql": jql,
                "maxResults": max_results,
                "fields": [
                    "summary", "status", "assignee",
                    "priority", "created", "updated",
                    "issuetype", "description",
                ],
            },
        )
        response.raise_for_status()
        return response.json().get("issues", [])

    async def create_issue(
        self,
        project_key: str,
        summary: str,
        description: str,
        issue_type: str = "Task",
    ) -> dict:
        response = await self.client.post(
            "/issue",
            json={
                "fields": {
                    "project": {"key": project_key},
                    "summary": summary,
                    "description": {
                        "type": "doc",
                        "version": 1,
                        "content": [
                            {
                                "type": "paragraph",
                                "content": [
                                    {"type": "text", "text": description}
                                ],
                            }
                        ],
                    },
                    "issuetype": {"name": issue_type},
                },
            },
        )
        response.raise_for_status()
        return response.json()

    async def add_comment(self, issue_key: str, body: str) -> dict:
        response = await self.client.post(
            f"/issue/{issue_key}/comment",
            json={
                "body": {
                    "type": "doc",
                    "version": 1,
                    "content": [
                        {
                            "type": "paragraph",
                            "content": [
                                {"type": "text", "text": body}
                            ],
                        }
                    ],
                },
            },
        )
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self.client.aclose()


# ---------------------------------------------------------------------------
# Lifespan
# ---------------------------------------------------------------------------

@dataclass
class AppContext:
    jira: JiraClient


@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    config = JiraConfig(
        base_url=os.environ["JIRA_BASE_URL"],
        email=os.environ["JIRA_EMAIL"],
        api_token=os.environ["JIRA_API_TOKEN"],
    )
    jira = JiraClient(config)
    try:
        yield AppContext(jira=jira)
    finally:
        await jira.close()


# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------

def handle_errors(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            error_map = {
                401: "Ошибка аутентификации. Проверьте JIRA_API_TOKEN.",
                403: "Нет доступа. Проверьте права учётной записи.",
                404: "Ресурс не найден в Jira.",
                429: "Превышен лимит запросов. Подождите и повторите.",
            }
            msg = error_map.get(status, f"Ошибка Jira API: {status}")
            logger.error(f"HTTP {status} in {func.__name__}: {e}")
            return msg
        except httpx.ConnectError:
            return "Не удалось подключиться к Jira. Проверьте JIRA_BASE_URL."
        except httpx.TimeoutException:
            return "Таймаут при запросе к Jira."
        except Exception as e:
            logger.exception(f"Unexpected error in {func.__name__}")
            return f"Внутренняя ошибка: {type(e).__name__}"
    return wrapper


# ---------------------------------------------------------------------------
# Server
# ---------------------------------------------------------------------------

mcp = FastMCP("Jira MCP Server", lifespan=app_lifespan)


def _get_jira(ctx: Context) -> JiraClient:
    return ctx.lifespan_context.jira


def _extract_text(adf: dict) -> str:
    if not adf or not isinstance(adf, dict):
        return "Нет описания"
    texts = []
    for block in adf.get("content", []):
        for inline in block.get("content", []):
            if inline.get("type") == "text":
                texts.append(inline["text"])
    return "\n".join(texts) if texts else "Нет описания"


# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------

@mcp.tool()
@handle_errors
async def get_issue(issue_key: str, ctx: Context) -> str:
    """Получить данные тикета Jira по ключу.

    Args:
        issue_key: Ключ тикета, например PROJ-123
    """
    if not re.match(r"^[A-Z][A-Z0-9_]+-\d+$", issue_key):
        return f"Некорректный формат ключа: {issue_key}. Ожидается PROJ-123."

    jira = _get_jira(ctx)
    data = await jira.get_issue(issue_key)
    fields = data["fields"]
    return (
        f"**{data['key']}**: {fields['summary']}\n"
        f"Статус: {fields['status']['name']}\n"
        f"Тип: {fields['issuetype']['name']}\n"
        f"Приоритет: {fields.get('priority', {}).get('name', 'Не указан')}\n"
        f"Назначен: {fields.get('assignee', {}).get('displayName', 'Не назначен')}\n"
        f"Создан: {fields['created']}\n"
        f"Обновлён: {fields['updated']}\n\n"
        f"Описание:\n{_extract_text(fields.get('description', {}))}"
    )


@mcp.tool()
@handle_errors
async def search_issues(
    jql: str, max_results: int = 20, ctx: Context = None
) -> str:
    """Поиск тикетов Jira по JQL-запросу.

    Args:
        jql: JQL-запрос, например 'project = PROJ AND status = "In Progress"'
        max_results: Максимальное количество результатов (по умолчанию 20)
    """
    jira = _get_jira(ctx)
    issues = await jira.search_issues(jql, max_results)

    if not issues:
        return "Тикеты не найдены."

    lines = []
    for issue in issues:
        fields = issue["fields"]
        assignee = fields.get("assignee")
        assignee_name = assignee.get("displayName", "—") if assignee else "—"
        lines.append(
            f"- **{issue['key']}**: {fields['summary']} "
            f"[{fields['status']['name']}] → {assignee_name}"
        )
    return f"Найдено {len(issues)} тикетов:\n\n" + "\n".join(lines)


@mcp.tool()
@handle_errors
async def create_issue(
    project_key: str,
    summary: str,
    description: str = "",
    issue_type: str = "Task",
    ctx: Context = None,
) -> str:
    """Создать новый тикет в Jira.

    Args:
        project_key: Ключ проекта, например PROJ
        summary: Заголовок тикета
        description: Описание тикета
        issue_type: Тип — Task, Bug, Story (по умолчанию Task)
    """
    jira = _get_jira(ctx)
    result = await jira.create_issue(
        project_key, summary, description, issue_type
    )
    return f"Создан тикет **{result['key']}**: {summary}"


@mcp.tool()
@handle_errors
async def add_comment(
    issue_key: str, body: str, ctx: Context = None
) -> str:
    """Добавить комментарий к тикету Jira.

    Args:
        issue_key: Ключ тикета, например PROJ-123
        body: Текст комментария
    """
    jira = _get_jira(ctx)
    await jira.add_comment(issue_key, body)
    return f"Комментарий добавлен к {issue_key}."


# ---------------------------------------------------------------------------
# Resources
# ---------------------------------------------------------------------------

@mcp.resource("jira://projects")
async def list_projects(ctx: Context) -> str:
    """Список доступных проектов Jira"""
    jira = _get_jira(ctx)
    response = await jira.client.get("/project")
    response.raise_for_status()
    projects = response.json()
    lines = [f"- **{p['key']}**: {p['name']}" for p in projects]
    return "\n".join(lines)


@mcp.resource("jira://issue/{issue_key}")
async def get_issue_resource(issue_key: str, ctx: Context) -> str:
    """Данные тикета для контекста AI"""
    jira = _get_jira(ctx)
    data = await jira.get_issue(issue_key)
    fields = data["fields"]
    return (
        f"# {data['key']}: {fields['summary']}\n\n"
        f"**Статус:** {fields['status']['name']}\n"
        f"**Тип:** {fields['issuetype']['name']}\n"
        f"**Приоритет:** {fields.get('priority', {}).get('name', 'N/A')}\n"
        f"**Назначен:** {fields.get('assignee', {}).get('displayName', 'Не назначен')}\n\n"
        f"**Описание:**\n{_extract_text(fields.get('description', {}))}"
    )


# ---------------------------------------------------------------------------
# Prompts
# ---------------------------------------------------------------------------

@mcp.prompt()
def triage_issue(issue_key: str) -> str:
    """Проанализировать тикет и предложить приоритет и назначение."""
    return (
        f"Проанализируй тикет {issue_key} в Jira.\n\n"
        f"1. Прочитай данные тикета через get_issue\n"
        f"2. Определи приоритет на основе описания и типа\n"
        f"3. Предложи, кому назначить, исходя из текущей загрузки команды\n"
        f"4. Оцени трудоёмкость в story points\n\n"
        f"Верни структурированный ответ с обоснованием."
    )


@mcp.prompt()
def sprint_review() -> str:
    """Сформировать отчёт по текущему спринту"""
    return (
        "Сформируй отчёт по текущему спринту:\n\n"
        "1. Найди все тикеты текущего спринта через search_issues\n"
        "2. Сгруппируй по статусам: Done, In Progress, To Do\n"
        "3. Посчитай процент выполнения\n"
        "4. Выдели заблокированные тикеты\n"
        "5. Сформулируй риски для завершения спринта\n\n"
        "Формат: markdown-таблица + текстовый анализ рисков."
    )


@mcp.prompt()
def release_notes(version: str) -> list[Message]:
    """Сгенерировать release notes для версии."""
    return [
        Message(
            f"Сгенерируй release notes для версии {version}.\n\n"
            f"Найди все тикеты с fixVersion = {version} через search_issues "
            f'с JQL: fixVersion = "{version}" AND status = Done\n\n'
            f"Сгруппируй: Features, Bug Fixes, Improvements.\n"
            f"Формат: markdown, готовый для публикации."
        ),
        Message(
            f"Формирую release notes для v{version}. "
            f"Ищу закрытые тикеты этой версии.",
            role="assistant",
        ),
    ]


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main():
    mcp.run()


if __name__ == "__main__":
    main()

Итого

MCP сервер на Python — это:

FastMCP — высокоуровневый API, минимум бойлерплейта
@mcp.tool() — регистрация инструментов с автогенерацией JSON Schema из типов
@mcp.resource() — данные для чтения, URI templates для динамических ресурсов
@mcp.prompt() — шаблоны промптов, одно- и мультитёрновые
Lifespan — инициализация/очистка ресурсов при старте/остановке
Error handling — текстовые ответы вместо исключений, декоратор для унификации
Тестирование — Client(transport=mcp) из пакета fastmcp для in-process тестов, MCP Inspector для ручной проверки
Публикация — pip-пакет, Docker, uvx

От пустого файла до production-сервера — один рабочий день. Большая часть времени уходит на бизнес-логику (работа с API сервиса), а не на инфраструктуру MCP. SDK берёт на себя транспорт, сериализацию, Schema generation и протокол.

Документация MCP SDK: modelcontextprotocol.io. Репозиторий Python SDK: github.com/modelcontextprotocol/python-sdk.