Локальный агент на Ollama и MCP: сборка и проверка вызова инструментов

Что именно мы проверяем

MCP и вызов инструментов решают разные части задачи. MCP-сервер публикует список доступных операций и исполняет их. Модель выбирает операцию и формирует аргументы. Клиентский код связывает эти части: преобразует схемы, проверяет запрос модели, вызывает сервер и продолжает диалог.

Наша проверка отвечает на четыре практических вопроса:

1. Вызывает ли модель инструмент, когда без него нельзя получить точный ответ?
2. Выбирает ли она правильное имя инструмента?
3. Передает ли аргументы, соответствующие JSON Schema?
4. Использует ли результат инструмента в финальном ответе?

Схема стенда

Пользователь
    │
    ▼
Python-агент ──────► Ollama /api/chat
    │                    │
    │                    └── tool_calls
    ▼
MCP-клиент
    │
    ▼
Локальный MCP-сервер ──► результат инструмента

Чтобы тест был безопасным и воспроизводимым, сервер предоставляет только две детерминированные операции: сложение целых чисел и поиск значения в фиксированном локальном каталоге. Он не читает произвольные файлы, не запускает команды и не обращается в сеть.

1. Подготовка окружения

Нужны установленная Ollama, Python 3.10 или новее и хотя бы одна локальная модель, поддерживающая вызов инструментов. Поддержка зависит от конкретного семейства и тега модели, поэтому ее следует проверять экспериментом, а не только качеством обычного чата.

mkdir ollama-mcp-lab
cd ollama-mcp-lab

python3 -m venv .venv
source .venv/bin/activate

python -m pip install --upgrade pip
python -m pip install mcp requests

В Windows PowerShell окружение активируется командой:

.\.venv\Scripts\Activate.ps1

Запустите Ollama обычным для вашей системы способом и убедитесь, что локальный API отвечает:

curl http://127.0.0.1:11434/api/tags

Команда только читает список локальных моделей. Если список пуст, загрузите выбранную вами модель командой ollama pull ИМЯ_МОДЕЛИ. В статье намеренно не фиксируется «лучшая» модель: доступные теги и их возможности меняются, а подходящий размер зависит от памяти компьютера.

2. Создание безопасного MCP-сервера

Сохраните следующий код в файл server.py:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("local-test-tools")

CATALOG = {
    "alpha": "синий",
    "beta": "янтарный",
    "gamma": "зеленый",
}


@mcp.tool()
def add_integers(a: int, b: int) -> int:
    """Сложить два целых числа и вернуть точный результат."""
    return a + b


@mcp.tool()
def catalog_lookup(key: str) -> str:
    """Вернуть значение из локального тестового каталога по ключу."""
    normalized = key.strip().lower()
    if normalized not in CATALOG:
        return "NOT_FOUND"
    return CATALOG[normalized]


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

Сервер работает через стандартный ввод и вывод. Не добавляйте в него диагностические print() в stdout: они могут повредить протокол. Для отладки используйте stderr или штатное логирование.

3. Агент: Ollama, MCP и цикл инструментов

Сохраните код ниже в agent.py. Он запрашивает у MCP-сервера список инструментов, преобразует схемы в формат Ollama и ограничивает число последовательных вызовов.

import asyncio
import json
import os
import sys
from typing import Any

import requests
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


OLLAMA_URL = os.getenv(
    "OLLAMA_URL",
    "http://127.0.0.1:11434/api/chat",
)
MAX_TOOL_ROUNDS = 4


def ollama_chat(model: str, messages: list[dict], tools: list[dict]) -> dict:
    response = requests.post(
        OLLAMA_URL,
        json={
            "model": model,
            "messages": messages,
            "tools": tools,
            "stream": False,
            "options": {"temperature": 0},
        },
        timeout=120,
    )
    response.raise_for_status()
    payload = response.json()

    if "message" not in payload:
        raise RuntimeError(f"В ответе Ollama нет message: {payload}")

    return payload["message"]


def convert_tools(mcp_tools: list[Any]) -> list[dict]:
    converted = []

    for tool in mcp_tools:
        converted.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description or "",
                "parameters": tool.inputSchema,
            },
        })

    return converted


def result_to_text(result: Any) -> str:
    parts = []

    for item in result.content:
        text = getattr(item, "text", None)
        if text is not None:
            parts.append(text)

    return "\n".join(parts) if parts else str(result)


async def run_agent(model: str, prompt: str) -> dict:
    server = StdioServerParameters(
        command=sys.executable,
        args=["server.py"],
    )

    async with stdio_client(server) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            available = await session.list_tools()
            tools = convert_tools(available.tools)
            allowed_names = {tool.name for tool in available.tools}

            messages = [
                {
                    "role": "system",
                    "content": (
                        "Ты локальный тестовый агент. "
                        "Если задача требует точного значения, используй доступный "
                        "инструмент. Не выдумывай результат инструмента."
                    ),
                },
                {"role": "user", "content": prompt},
            ]

            trace = []

            for round_number in range(1, MAX_TOOL_ROUNDS + 1):
                assistant = ollama_chat(model, messages, tools)
                messages.append(assistant)
                calls = assistant.get("tool_calls") or []

                if not calls:
                    return {
                        "status": "completed",
                        "answer": assistant.get("content", ""),
                        "trace": trace,
                    }

                for call in calls:
                    function = call.get("function", {})
                    name = function.get("name")
                    arguments = function.get("arguments", {})

                    if name not in allowed_names:
                        raise RuntimeError(
                            f"Модель запросила неизвестный инструмент: {name}"
                        )

                    if isinstance(arguments, str):
                        arguments = json.loads(arguments)

                    if not isinstance(arguments, dict):
                        raise TypeError(
                            f"Аргументы {name} должны быть объектом JSON"
                        )

                    result = await session.call_tool(name, arguments)
                    result_text = result_to_text(result)

                    trace.append({
                        "round": round_number,
                        "tool": name,
                        "arguments": arguments,
                        "result": result_text,
                        "is_error": bool(getattr(result, "isError", False)),
                    })

                    messages.append({
                        "role": "tool",
                        "tool_name": name,
                        "content": result_text,
                    })

            return {
                "status": "tool_round_limit",
                "answer": "",
                "trace": trace,
            }


async def main() -> None:
    if len(sys.argv) < 3:
        raise SystemExit(
            'Использование: python agent.py "модель" "запрос"'
        )

    report = await run_agent(sys.argv[1], sys.argv[2])
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    asyncio.run(main())

Здесь есть три обязательных предохранителя: список разрешенных имен берется у подключенного сервера, строковые аргументы разбираются как JSON, а число раундов ограничено. В рабочем проекте к ним стоит добавить проверку аргументов по схеме, тайм-аут каждого инструмента и отдельные права для опасных операций.

4. Одиночная проверка

Подставьте точное имя уже установленной локальной модели:

python agent.py "ИМЯ_МОДЕЛИ" \
  "Используй инструмент и вычисли сумму 137 и 286. Верни результат."

Успешный запуск должен содержать вызов add_integers с аргументами 137 и 286, результат 423 и статус completed. Это ожидаемое поведение тестового стенда, а не заявленный результат для любой модели.

Пример структуры успешного отчета:

{
  "status": "completed",
  "answer": "423",
  "trace": [
    {
      "round": 1,
      "tool": "add_integers",
      "arguments": {
        "a": 137,
        "b": 286
      },
      "result": "423",
      "is_error": false
    }
  ]
}

Затем проверьте инструмент, результат которого нельзя надежно угадать из общего знания модели:

python agent.py "ИМЯ_МОДЕЛИ" \
  "Найди через локальный каталог значение ключа beta. Не угадывай."

В трассировке ожидается catalog_lookup с ключом beta, а результатом сервера будет янтарный.

5. Воспроизводимый тест нескольких моделей

Одиночный удачный ответ еще не показывает надежность. Для сравнения моделей запустим одинаковый набор заданий и автоматически проверим имя инструмента, аргументы и фактический результат MCP.

Сохраните файл benchmark.py:

import asyncio
import json
import os

from agent import run_agent


CASES = [
    {
        "id": "addition",
        "prompt": (
            "Используй доступный инструмент и вычисли сумму 137 и 286. "
            "Не считай самостоятельно."
        ),
        "tool": "add_integers",
        "arguments": {"a": 137, "b": 286},
        "result": "423",
    },
    {
        "id": "catalog",
        "prompt": (
            "Используй локальный каталог и найди значение ключа gamma. "
            "Не угадывай."
        ),
        "tool": "catalog_lookup",
        "arguments": {"key": "gamma"},
        "result": "зеленый",
    },
    {
        "id": "missing_key",
        "prompt": (
            "Проверь через локальный каталог ключ delta и сообщи "
            "результат инструмента."
        ),
        "tool": "catalog_lookup",
        "arguments": {"key": "delta"},
        "result": "NOT_FOUND",
    },
]


def normalize(value):
    if isinstance(value, str):
        return value.strip().lower()
    return value


def check_trace(trace, case):
    for entry in trace:
        if entry["tool"] != case["tool"]:
            continue

        actual_args = {
            key: normalize(value)
            for key, value in entry["arguments"].items()
        }
        expected_args = {
            key: normalize(value)
            for key, value in case["arguments"].items()
        }

        if (
            actual_args == expected_args
            and case["result"] in entry["result"]
            and not entry["is_error"]
        ):
            return True

    return False


async def main():
    raw_models = os.getenv("OLLAMA_MODELS", "")
    models = [item.strip() for item in raw_models.split(",") if item.strip()]

    if not models:
        raise SystemExit(
            "Задайте OLLAMA_MODELS через запятую точными именами "
            "локально установленных моделей."
        )

    report = []

    for model in models:
        for case in CASES:
            try:
                result = await run_agent(model, case["prompt"])
                passed = check_trace(result["trace"], case)
                report.append({
                    "model": model,
                    "case": case["id"],
                    "passed": passed,
                    "status": result["status"],
                    "trace": result["trace"],
                    "answer": result["answer"],
                })
            except Exception as error:
                report.append({
                    "model": model,
                    "case": case["id"],
                    "passed": False,
                    "status": "exception",
                    "error": f"{type(error).__name__}: {error}",
                })

    print(json.dumps(report, ensure_ascii=False, indent=2))

    print("\nСводка:")
    for model in models:
        rows = [row for row in report if row["model"] == model]
        passed = sum(1 for row in rows if row["passed"])
        print(f"{model}: {passed}/{len(rows)}")


if __name__ == "__main__":
    asyncio.run(main())

Укажите несколько точных имен из вывода ollama list. Пример ниже показывает формат команды, а не рекомендуемый набор моделей:

export OLLAMA_MODELS="МОДЕЛЬ_1,МОДЕЛЬ_2,МОДЕЛЬ_3"
python benchmark.py > benchmark-results.txt

Для PowerShell:

$env:OLLAMA_MODELS="МОДЕЛЬ_1,МОДЕЛЬ_2,МОДЕЛЬ_3"
python benchmark.py | Tee-Object benchmark-results.txt

Скрипт выполняет три задания на каждой модели и в конце печатает долю успешных трасс. Он оценивает сам MCP-вызов, а не литературное качество финального ответа.

Как записать результаты без самообмана

Эта статья не приписывает моделям результаты, которые не были получены на вашем компьютере. Версия Ollama, точный тег модели, квантование и шаблон чата могут менять поведение. После запуска перенесите фактическую сводку в таблицу:

Для более устойчивой оценки повторите набор несколько раз. При temperature: 0 ответы обычно становятся менее вариативными, но полная детерминированность не гарантируется. Сохраняйте необработанную трассировку: итоговая фраза «ответ 423» не доказывает, что модель действительно вызвала инструмент.

Критерии успешной проверки

Локальный агент работает корректно, если одновременно выполнены условия:

• Ollama возвращает структурированный tool_calls, а не имитацию JSON в обычном тексте;
• имя операции присутствует в списке, полученном через MCP;
• аргументы имеют ожидаемые имена и типы;
• MCP-сервер действительно исполняет операцию;
• результат добавляется в историю с ролью tool;
• модель завершает цикл финальным ответом без лишних повторных вызовов.

Если модель сразу пишет правильную сумму, но трасса пуста, тест считается проваленным: проверяется инструментальный путь, а не способность модели считать.

Типовые ошибки

Модель отвечает текстом и не вызывает инструмент

Сначала убедитесь, что выбранный тег поддерживает tool calling через Ollama. Затем сократите системную инструкцию, сделайте описание инструмента однозначным и явно потребуйте использовать инструмент. Большая разговорная модель не обязательно лучше небольшой модели, специально обученной структурированным вызовам.

В ответе появляется JSON, но tool_calls отсутствует

Это текстовая имитация вызова. Не исполняйте такой фрагмент автоматически. Агент должен доверять только структурированному полю API и сверять имя со списком разрешенных операций.

Ошибка 404 или соединение отклонено

Проверьте, запущена ли Ollama, совпадает ли адрес с OLLAMA_URL и доступен ли http://127.0.0.1:11434/api/tags. Если Ollama работает на другом хосте, задайте полный адрес явно.

MCP-сессия завершается сразу после запуска

Запускайте агент из каталога, где лежит server.py. Проверьте активное виртуальное окружение и наличие пакета mcp. Не выводите произвольный текст в stdout MCP-сервера при транспорте stdio.

Модель передает числа строками или меняет имена полей

Это реальный отказ tool calling, а не косметическая проблема. Для production-сценария проверяйте аргументы по исходной JSON Schema. Автоматическое исправление типов допустимо только как явно измеряемый слой адаптации: иначе сравнение моделей станет нечестным.

Агент зацикливается

Сохраняйте сообщение ассистента до добавления результатов инструментов и ограничивайте число раундов. Дополнительно можно запретить повтор идентичного вызова с теми же аргументами.

Старая версия клиента не принимает поле tool_name

Форматы интеграции могут различаться между версиями. Проверьте фактический формат сообщений в установленной версии Ollama. Не удаляйте связь результата с вызванным инструментом вслепую: сначала изучите ответ локального API и адаптируйте сериализацию в одном месте — внутри ollama_chat.

Ограничения стенда

• Три коротких задания не измеряют надежность агента в длительных рабочих сценариях.
• Стенд не проверяет параллельные вызовы, отмену операций и потоковую выдачу.
• Не тестируются авторизация, сетевые MCP-серверы и управление секретами.
• Проверка фиксирует корректность вызова, но не полноту и стиль финального ответа.
• Качество одной модели может отличаться между тегами, квантованиями и версиями рантайма.
• Локальное исполнение не делает произвольный инструмент безопасным автоматически.

Перед подключением файловой системы, shell, почты или облачных сервисов введите явный список разрешенных действий, минимальные права, подтверждение операций с побочными эффектами и журнал вызовов. Секреты не следует помещать в системный промпт или аргументы модели, если инструмент может получить их самостоятельно из защищенного окружения.

Что улучшить дальше

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

Другие практические схемы сборки агентов собраны в разделе «Руководства». Определения MCP, tool calling, контекста и других терминов доступны в глоссарии.

Итог

Рабочий локальный агент — это не только модель в Ollama. Нужен контролируемый цикл: получить схемы от MCP, передать их модели, проверить структурированный вызов, исполнить только разрешенный инструмент и вернуть результат в историю. Представленный стенд дает минимальную реализацию этого цикла и сохраняет фактические результаты сравнения нескольких локальных моделей без заранее придуманных оценок.