Локальный AI с Qwen3.6 и MCP: GitHub-ассистент без облаков

# Требуется Python 3.11+
python --version
python -m venv qwen-mcp-env
source qwen-mcp-env/bin/activate  # macOS / Linux
qwen-mcp-env\Scripts\activate    # Windows

# Основные пакеты
pip install \
    "openai>=1.30.0" \
    "qwen-agent>=0.0.10" \
    "mcp>=1.0.0" \
    "httpx>=0.27.0"

# Фреймворки для развёртывания (выберите один)
pip install "vllm>=0.19.0"   # NVIDIA GPU
pip install "sglang>=0.5.10" # NVIDIA GPU (быстрее заполняет длинный контекст)
pip install "ktransformers"   # CPU/гибрид

# Требуется Node.js 18+ для готовых MCP-серверов, запускаемых через npx
node --version

# Установка SGLang со всеми зависимостями
pip install "sglang[all]>=0.5.10"

# Запуск Qwen3.6-35B-A3B с включёнными парсерами рассуждений и вызовов инструментов.
# --reasoning-parser qwen3 корректно обрабатывает блоки <think>...</think>.
# --tool-call-parser qwen3_coder направляет результаты вызовов инструментов в нужный формат.
# --enable-prefix-caching критично для агентных нагрузок — включает повторное использование KV-кеша
# между шагами, что делает preserve_thinking эффективным на практике.
python -m sglang.launch_server \
    --model-path Qwen/Qwen3.6-35B-A3B \
    --host 0.0.0.0 \
    --port 30000 \
    --reasoning-parser qwen3 \
    --tool-call-parser qwen3_coder \
    --enable-prefix-caching \
    --tp 2  # параллелизм по тензорам на 2 GPU; удалите, если GPU один

pip install "vllm>=0.19.0"

# Эквивалент vLLM с теми же критичными флагами
vllm serve Qwen/Qwen3.6-35B-A3B \
    --host 0.0.0.0 \
    --port 8000 \
    --reasoning-parser qwen3 \
    --tool-call-parser qwen3_coder \
    --enable-prefix-caching-v2 \
    --tensor-parallel-size 2

ollama pull qwen2.5:7b
ollama serve
# API Ollama совместим с OpenAI по адресу http://localhost:11434/v1

# Проверка состояния — должен вернуть {"status": "ok"} или подобное
curl http://localhost:30000/health

# Тест эндпоинта чат-дополнений с простым запросом
curl http://localhost:30000/v1/chat/completions \
    -H "Content-Type: application/json" \
    -d '{
        "model": "Qwen/Qwen3.6-35B-A3B",
        "messages": [{"role": "user", "content": "Ответь: готов"}],
        "max_tokens": 10
    }'

# Установите персональный токен доступа GitHub
# Требуется GitHub MCP-серверу для вызовов API
export GITHUB_TOKEN=ghp_your_token_here

# Готовые MCP-серверы устанавливаются через npx — отдельный шаг установки не нужен.
# npx сделает это при первом использовании, когда агент запустит серверы.
# Проверьте наличие npx: npx --version

mkdir qwen-github-agent
cd qwen-github-agent

# github_agent_qwenagent.py
# Предварительно: pip install qwen-agent openai
# Должны быть установлены npm / npx для MCP-серверов
# Переменная окружения GITHUB_TOKEN должна быть задана
# Локальный сервер инференса должен быть запущен (см. предыдущий раздел)
#
# Запуск:
#   python github_agent_qwenagent.py

from qwen_agent.agents import Assistant

# ── Конфигурация сервера ─────────────────────────────────────────────────────
# Укажите адрес вашего локального сервера.
# Измените base_url в соответствии с тем, какой сервер вы запустили:
#   SGLang: http://localhost:30000/v1
#   vLLM:   http://localhost:8000/v1
#   Ollama: http://localhost:11434/v1
LLM_CONFIG = {
    "model": "Qwen/Qwen3.6-35B-A3B",
    "model_server": "http://localhost:30000/v1",
    "api_key": "EMPTY",  # Локальным серверам не нужен настоящий ключ

    # Параметры сэмплирования для режима рассуждений (из официальных рекомендаций)
    "generate_cfg": {
        "temperature": 0.6,
        "top_p": 0.95,
        "top_k": 20,
        "min_p": 0.0,
        "thought_in_history": True,  # Это флаг preserve_thinking в Qwen-Agent
    },
}

# ── Конфигурация MCP-серверов ────────────────────────────────────────────────
# Ключи словаря — имена серверов, значения — команды запуска через stdio.
# Qwen-Agent запускает каждый сервер как подпроцесс и управляет сессиями MCP.
MCP_SERVERS = {
    "mcpServers": {
        "filesystem": {
            "command": "npx",
            "args": [
                "-y",
                "@modelcontextprotocol/server-filesystem",
                # Даём агенту доступ к текущей рабочей директории
                # В production ограничьте путь конкретным репозиторием
                "."
            ]
        },
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {
                # GitHub MCP-сервер читает эту переменную для аутентификации API
                "GITHUB_TOKEN": "${GITHUB_TOKEN}"
            }
        },
    }
}

# ── Системный промпт ─────────────────────────────────────────────────────────
SYSTEM_PROMPT = """Вы — старший инженер-программист с полным доступом к
GitHub-репозиторию через инструменты MCP. Когда вам дают репозиторий и задачу:
1. Просматривайте открытые issue, чтобы понять, что нужно исправить.
2. Используйте файловые инструменты для чтения нужных исходных файлов и тестов.
3. Определяйте корневую причину на основе кода и описания issue.
4. Пишите точечное исправление — с минимальными изменениями,
   без рефакторинга, не связанного с ошибкой.
5. Создавайте pull request с чётким заголовком и описанием, ссылаясь на issue.
Всегда объясняйте свои рассуждения на каждом шаге. Думайте о граничных случаях,
прежде чем писать код. Если вы не уверены в назначении файла,
прочитайте его, прежде чем модифицировать."""

# ── Настройка агента ─────────────────────────────────────────────────────────
agent = Assistant(
    llm=LLM_CONFIG,
    name="GitHub Developer Assistant",
    description="Читает issue, исправляет ошибки, открывает pull request — локально через MCP.",
    system_message=SYSTEM_PROMPT,
    mcp_servers=MCP_SERVERS,
)

# ── Запуск агента ────────────────────────────────────────────────────────────
def run_agent(task: str):
    """
    Запускает агента с описанием задачи и выводит результат.
    Агент автоматически выполняет вызовы инструментов;
    Qwen-Agent зацикливает весь процесс, включая исполнение инструментов и вставку результатов.
    """
    messages = [{"role": "user", "content": task}]
    print(f"Задача: {task}\n{'─' * 70}")

    # run() возвращает генератор, выдающий промежуточные шаги
    # Каждое значение — список сообщений, представляющий текущее состояние диалога
    for response in agent.run(messages=messages):
        last = response[-1]
        role = last.get("role", "")
        content = last.get("content", "")

        if role == "assistant" and content:
            # Выделяем блок рассуждений для удобства чтения
            import re
            thinking = re.search(r"<think>(.*?)</think>", content, re.DOTALL)
            if thinking:
                print(f"[thinking] {thinking.group(1).strip()[:200]}...")
            clean = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL).strip()
            if clean:
                print(f"[agent] {clean}")
        elif role == "tool":
            tool_name = last.get("name", "unknown_tool")
            print(f"[tool:{tool_name}] получен результат")

if __name__ == "__main__":
    run_agent(
        "В репозитории myorg/my-api-project найдите открытую issue о том, "
        "что эндпоинт логина возвращает 200 для невалидных токенов. "
        "Прочитайте соответствующий код и тесты, исправьте ошибку и откройте pull request."
    )

python github_agent_qwenagent.py

# github_agent_raw.py
# Предварительно: pip install mcp openai httpx
# Должны быть установлены переменная GITHUB_TOKEN и локальный сервер
#
# Запуск:
#   python github_agent_raw.py

import asyncio
import json
import os
import re

from openai import AsyncOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# ── Клиент локального сервера ───────────────────────────────────────────────
client = AsyncOpenAI(
    base_url="http://localhost:30000/v1",
    api_key="EMPTY",
)
MODEL = "Qwen/Qwen3.6-35B-A3B"

# ── Обработка ответов ────────────────────────────────────────────────────────
def strip_thinking(text: str) -> str:
    """Удаляет блоки <think>...</think>. Используется, когда нужно только действие."""
    return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()

def extract_thinking(text: str) -> str:
    """Извлекает содержимое блока thinking для логирования."""
    m = re.search(r"<think>(.*?)</think>", text, re.DOTALL)
    return m.group(1).strip() if m else ""

def process_response(response, preserve_thinking: bool = True) -> dict:
    """
    Обрабатывает ответ чат-дополнения от Qwen3.6.
    Поддерживает два формата вывода:
    1. Вызов инструмента через поля function_call / tool_calls API
       (когда активен --tool-call-parser).
    2. Вызов инструмента, встроенный в содержимое сообщения как <tool_call>JSON</tool_call>.

    Аргументы:
        response: OpenAI-совместимый ответ дополнения.
        preserve_thinking: Если True, сохраняет содержимое thinking в выводе
            для выгоды KV-кеша на следующем шаге.

    Возвращает:
        dict с ключами thinking, tool_calls, final_answer, has_tool_calls, is_terminal.
    """
    choice = response.choices[0]
    message = choice.message

    # Путь 1: вызовы инструментов в структурированном поле (предпочтительный — требует флаг tool-call-parser)
    if message.tool_calls:
        tool_calls = [
            {
                "name": tc.function.name,
                "arguments": json.loads(tc.function.arguments),
                "call_id": tc.id,
            }
            for tc in message.tool_calls
        ]
        thinking = extract_thinking(message.content or "")
        return {
            "thinking": thinking if preserve_thinking else "",
            "tool_calls": tool_calls,
            "final_answer": "",
            "has_tool_calls": True,
            "is_terminal": False,
        }

    # Путь 2: вызовы инструментов в тексте сообщения (запасной)
    content = message.content or ""
    tag_matches = re.findall(r"<tool_call>(.*?)</tool_call>", content, re.DOTALL)
    tool_calls = []
    for m in tag_matches:
        try:
            tool_calls.append(json.loads(m.strip()))
        except json.JSONDecodeError:
            pass

    thinking = extract_thinking(content)
    final_answer = re.sub(r"<think>.*?</think>", "", content, flags=re.DOTALL)
    final_answer = re.sub(r"<tool_call>.*?</tool_call>", "", final_answer, flags=re.DOTALL).strip()

    return {
        "thinking": thinking if preserve_thinking else "",
        "tool_calls": tool_calls,
        "final_answer": final_answer,
        "has_tool_calls": len(tool_calls) > 0,
        "is_terminal": len(tool_calls) == 0 and bool(final_answer),
    }

# ── Основной цикл агента ────────────────────────────────────────────────────
async def run_github_agent(task: str, repo: str, max_turns: int = 20):
    """
    Запускает агента-ассистента GitHub.
    Подключается к файловому и GitHub MCP-серверам, получает их инструменты
    и выполняет цикл агента Qwen3.6, пока задача не будет выполнена или не достигнут лимит ходов.
    """
    # Запускаем оба MCP-сервера и устанавливаем сессии
    fs_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "."],
    )
    gh_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={**os.environ, "GITHUB_TOKEN": os.environ.get("GITHUB_TOKEN", "")},
    )

    async with stdio_client(fs_params) as (fs_read, fs_write), \
            ClientSession(fs_read, fs_write) as fs_session, \
            stdio_client(gh_params) as (gh_read, gh_write), \
            ClientSession(gh_read, gh_write) as gh_session:

        # Инициализируем обе сессии
        await fs_session.initialize()
        await gh_session.initialize()

        # Получаем списки всех доступных инструментов
        fs_tools_result = await fs_session.list_tools()
        gh_tools_result = await gh_session.list_tools()

        # Формируем список инструментов в формате OpenAI и словарь маршрутизации
        all_tools = []
        tool_to_session = {}  # Сопоставляет имя инструмента с его MCP-сессией

        for tool in fs_tools_result.tools:
            all_tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                }
            })
            tool_to_session[tool.name] = fs_session

        for tool in gh_tools_result.tools:
            all_tools.append({
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                }
            })
            tool_to_session[tool.name] = gh_session

        print(f"Доступно инструментов: {len(all_tools)} "
              f"({len(fs_tools_result.tools)} файловых, {len(gh_tools_result.tools)} GitHub)")

        # Формируем историю диалога
        system_prompt = f"""Вы — старший инженер-программист с доступом к репозиторию {repo}.
Используйте доступные инструменты для исследования проблем, чтения кода, написания исправлений
и создания pull request. Думайте пошагово. Читайте перед тем, как модифицировать.
Только минимальные изменения."""

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": task},
        ]

        # ── Цикл агента ─────────────────────────────────────────────────────
        for turn in range(max_turns):
            print(f"\n[Ход {turn + 1}]")

            # Вызов модели
            response = await client.chat.completions.create(
                model=MODEL,
                messages=messages,
                tools=all_tools if all_tools else None,
                tool_choice="auto",
                # Параметры сэмплирования для режима рассуждений
                temperature=0.6,
                top_p=0.95,
                top_k=20,
                min_p=0.0,
                max_tokens=4096,
                extra_body={
                    # preserve_thinking сохраняет контекст рассуждений между ходами
                    # для эффективности KV-кеша в длинных агентных сессиях
                    "preserve_thinking": True,
                }
            )

            result = process_response(response, preserve_thinking=True)
            if result["thinking"]:
                print(f"[thinking] {result['thinking'][:200]}...")

            # Терминальное состояние — агент выдал финальный ответ
            if result["is_terminal"]:
                print(f"\n[ГОТОВО]\n{result['final_answer']}")
                return result["final_answer"]

            # Состояние вызова инструментов — выполняем каждый и вставляем результаты
            if result["has_tool_calls"]:
                # Добавляем сообщение ассистента с вызовами инструментов в историю
                messages.append({
                    "role": "assistant",
                    "content": response.choices[0].message.content or "",
                    "tool_calls": response.choices[0].message.tool_calls or [],
                })

                for call in result["tool_calls"]:
                    tool_name = call["name"]
                    tool_args = call.get("arguments", {})
                    call_id = call.get("call_id", "")
                    print(f"[tool] {tool_name}({json.dumps(tool_args)[:80]}...)")

                    session = tool_to_session.get(tool_name)
                    if not session:
                        result_content = f"Ошибка: инструмент '{tool_name}' не найден"
                    else:
                        try:
                            tool_result = await session.call_tool(tool_name, tool_args)
                            result_content = str(tool_result.content)
                            # Обрезаем очень длинные результаты для экономии контекста
                            if len(result_content) > 12000:
                                result_content = result_content[:12000] + "\n...[обрезано]"
                        except Exception as e:
                            result_content = f"Ошибка: {e}"

                    print(f"[result] {result_content[:150]}...")
                    messages.append({
                        "role": "tool",
                        "content": result_content,
                        "tool_call_id": call_id,
                        "name": tool_name,
                    })

        print(f"[ПРЕДУПРЕЖДЕНИЕ] Достигнут лимит ходов ({max_turns}) без терминального состояния")

# ── Точка входа ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
    asyncio.run(run_github_agent(
        task=(
            "Найдите открытую issue о том, что эндпоинт логина возвращает 200 для невалидных токенов. "
            "Прочитайте src/auth.py и tests/test_auth.py, чтобы понять ошибку. "
            "Исправьте функцию verify_token и откройте pull request с вашими изменениями."
        ),
        repo="myorg/my-api-project",
    ))

python github_agent_raw.py

# code_quality_server.py
# Пользовательский MCP-сервер, предоставляющий инструменты контроля качества кода для Qwen3.6.
#
# Предварительно: pip install mcp ruff pytest
#
# Автономный запуск (для тестирования):
#   python code_quality_server.py
#
# Добавление в конфигурацию Qwen-Agent:
#   "code_quality": {
#       "command": "python",
#       "args": ["/absolute/path/to/code_quality_server.py"]
#   }

import asyncio
import json
import subprocess
import sys

from mcp.server.fastmcp import FastMCP

# FastMCP — высокоуровневый фреймворк для MCP-серверов, уменьшает количество шаблонного кода
mcp = FastMCP("code_quality")

@mcp.tool()
def run_linter(file_path: str, fix: bool = False) -> str:
    """
    Запускает ruff-линтер для Python-файла и возвращает структурированные результаты.
    Используйте перед модификацией файла, чтобы понять его текущее состояние качества,
    и после внесения изменений, чтобы убедиться, что исправление не добавило новых проблем.

    Аргументы:
        file_path: Абсолютный или относительный путь к файлу Python.
        fix: Если True, автоматически исправляет безопасные проблемы на месте.

    Возвращает:
        Строку JSON со списком проблем, их количеством и информацией об изменённых файлах.
    """
    cmd = ["python", "-m", "ruff", "check", file_path, "--output-format=json"]
    if fix:
        cmd.append("--fix")
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        # ruff возвращает код выхода 1, когда проблемы найдены — это не ошибка
        output = result.stdout or result.stderr

        # Парсим JSON-вывод ruff
        try:
            issues = json.loads(output) if output.strip() else []
        except json.JSONDecodeError:
            issues = []

        formatted = [
            {
                "line": issue.get("location", {}).get("row", 0),
                "col": issue.get("location", {}).get("column", 0),
                "code": issue.get("code", ""),
                "message": issue.get("message", ""),
                "fix_available": issue.get("fix") is not None,
            }
            for issue in issues if isinstance(issue, dict)
        ]

        return json.dumps({
            "file": file_path,
            "issues": formatted,
            "total_issues": len(formatted),
            "fixed": "автоисправление применено" if fix else "автоисправлений не было",
        }, indent=2)

    except subprocess.TimeoutExpired:
        return json.dumps({"error": "Линтер превысил время ожидания (30 с)", "file": file_path})
    except FileNotFoundError:
        return json.dumps({"error": "ruff не найден — установите: pip install ruff"})

@mcp.tool()
def run_tests(target: str, verbose: bool = False) -> str:
    """
    Запускает pytest для модуля или директории и возвращает структурированные результаты.
    Используйте после написания исправления, чтобы убедиться, что падавшие тесты проходят,
    а остальные не сломались.

    Аргументы:
        target: Путь к тестовому файлу или директории (например, tests/, tests/test_auth.py)
        verbose: Если True, включить полный вывод pytest.

    Возвращает:
        Строку JSON с количеством пройденных, упавших, ошибок, деталями падений и длительностью.
    """
    cmd = ["python", "-m", "pytest", target, "--json-report", "--json-report-file=-", "-q"]
    if verbose:
        cmd.append("-v")
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        output = result.stdout

        # Парсим вывод pytest-json-report, если он есть
        try:
            report = json.loads(output)
            summary = report.get("summary", {})
            failures = [
                {
                    "test": t["nodeid"],
                    "message": t.get("call", {}).get("longrepr", "")[:500],
                }
                for t in report.get("tests", []) if t.get("outcome") == "failed"
            ]
            return json.dumps({
                "target": target,
                "passed": summary.get("passed", 0),
                "failed": summary.get("failed", 0),
                "errors": summary.get("error", 0),
                "total": summary.get("total", 0),
                "duration": summary.get("duration", 0),
                "failures": failures,
                "stdout": result.stdout[:2000] if verbose else "",
            }, indent=2)
        except json.JSONDecodeError:
            # Запасной вариант: возвращаем сырой вывод, если JSON-отчёт не получен
            return json.dumps({
                "target": target,
                "stdout": result.stdout[:3000],
                "stderr": result.stderr[:1000],
                "exit_code": result.returncode,
            })
    except subprocess.TimeoutExpired:
        return json.dumps({"error": f"Тесты превысили время ожидания (120 с) для цели: {target}"})
    except FileNotFoundError:
        return json.dumps({"error": "pytest не найден — установите: pip install pytest"})

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

# В словарь MCP_SERVERS Qwen-Agent:
"code_quality": {
    "command": "python",
    "args": ["/absolute/path/to/code_quality_server.py"]
}

# В чистом SDK — добавьте третий StdioServerParameters:
cq_params = StdioServerParameters(
    command="python",
    args=["/absolute/path/to/code_quality_server.py"],
)

# Тестирование в режиме MCP Inspector
npx @modelcontextprotocol/inspector python code_quality_server.py
# Откроется браузерный интерфейс, где можно напрямую вызывать run_linter и run_tests

# Режим размышлений (планирование, отладка, сложные многофайловые задачи)
THINKING_PARAMS = {
    "temperature": 0.6,
    "top_p": 0.95,
    "top_k": 20,
    "min_p": 0.0,
}

# Режим без размышлений (механические циклы, быстрые проверки состояния)
# Передайте enable_thinking=False в шаблон чата или добавьте /no_think в системный промпт.
NON_THINKING_PARAMS = {
    "temperature": 0.7,
    "top_p": 0.8,
    "top_k": 20,
    "min_p": 0.0,
}

GitHub-ассистент из этой статьи — одно применение схемы. Та же архитектура — локальная модель, инструменты MCP и агентный цикл — работает для научного ассистента, который ищет в академических базах и составляет обзоры литературы, для DevOps-агента, читающего CloudWatch-логи и заводящего инциденты, или для агента пайплайнов данных, изучающего SQL-схемы, пишущего код трансформации и проверяющего выходные данные. Экосистема MCP стремительно растёт. Возможности локальных моделей уже здесь.