Sampling на практике

mkdir ci-reporter && cd ci-reporter
uv init
uv add "mcp[cli]"

import logging

from mcp.server import MCPServer

logger = logging.getLogger(__name__)
mcp = MCPServer("ci-reporter")

# Вместо реального CI — заглушка. В настоящем сервере здесь будет вызов API.
FAILURES: dict[int, str] = {
    4812: (
        "FAILED tests/test_checkout.py::test_apply_promo\n"
        "AssertionError: expected total 900, got 1000\n"
        "  at checkout/pricing.py:57 in apply_promo\n"
        "  promo 'SPRING10' loaded, discount=0"
    ),
    4813: (
        "FAILED tests/test_export.py::test_csv_header\n"
        "UnicodeEncodeError: 'ascii' codec can't encode character '\\u0451'\n"
        "  at export/csv_writer.py:23 in write_header"
    ),
}


def fetch_failure_log(build_id: int) -> str:
    log = FAILURES.get(build_id)
    if log is None:
        return "Сборка не найдена или завершилась успешно."
    return log


@mcp.tool()
def failure_log(build_id: int) -> str:
    """Вернуть сырой лог упавшего теста по номеру сборки."""
    logger.info("failure_log for build %s", build_id)
    return fetch_failure_log(build_id)


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

from typing import Annotated

from mcp.server.mcpserver import Resolve, Sample
from mcp.types import CreateMessageResult, SamplingMessage, TextContent


def summarize_failure(build_id: int) -> Sample:
    """Резолвер: собирает запрос к модели из лога сборки."""
    log = fetch_failure_log(build_id)
    prompt = (
        "Ниже лог упавшего теста. Опиши проблему для тикета в двух предложениях: "
        "что сломалось и в каком файле. Без вступлений.\n\n"
        f"{log}"
    )
    return Sample(
        [SamplingMessage(role="user", content=TextContent(type="text", text=prompt))],
        max_tokens=200,
        system_prompt="Ты пишешь краткие технические описания багов на русском языке.",
    )


@mcp.tool()
async def report_failure(
    build_id: int,
    summary: Annotated[CreateMessageResult, Resolve(summarize_failure)],
) -> str:
    """Подготовить описание упавшей сборки для тикета."""
    if summary.content.type != "text":
        return "Модель вернула не текст, описание не составлено."
    return f"Сборка {build_id}. {summary.content.text}"

import anyio

from mcp import Client
from mcp.client.session import ClientRequestContext
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent

from server import mcp


async def fake_model(
    context: ClientRequestContext,
    params: CreateMessageRequestParams,
) -> CreateMessageResult:
    """Играет роль модели хоста: печатает, что попросил сервер, и отвечает заглушкой."""
    first = params.messages[0].content
    print("--- сервер попросил модель ---")
    print("system_prompt:", params.system_prompt)
    print("max_tokens:", params.max_tokens)
    if first.type == "text":
        print("prompt начинается с:", first.text[:60].replace("\n", " "), "...")
    return CreateMessageResult(
        role="assistant",
        content=TextContent(
            type="text",
            text=(
                "Тест test_apply_promo падает: промокод SPRING10 загружается, "
                "но скидка не применяется, итог 1000 вместо 900. "
                "Проблема в checkout/pricing.py, функция apply_promo."
            ),
        ),
        model="fake-model",
    )


async def main() -> None:
    async with Client(mcp, sampling_callback=fake_model) as client:
        tools = await client.list_tools()
        print("инструменты:", [t.name for t in tools.tools])
        print("схема report_failure:", tools.tools[1].input_schema["properties"])

        result = await client.call_tool("report_failure", {"build_id": 4812})
        print("--- результат инструмента ---")
        print(result.content[0].text)


anyio.run(main)

инструменты: ['failure_log', 'report_failure']
схема report_failure: {'build_id': {'title': 'Build Id', 'type': 'integer'}}
--- сервер попросил модель ---
system_prompt: Ты пишешь краткие технические описания багов на русском языке.
max_tokens: 200
prompt начинается с: Ниже лог упавшего теста. Опиши проблему для тикета в двух ...
--- результат инструмента ---
Сборка 4812. Тест test_apply_promo падает: промокод SPRING10 загружается, но скидка не применяется, итог 1000 вместо 900. Проблема в checkout/pricing.py, функция apply_promo.

{
  "jsonrpc": "2.0",
  "id": 3,
  "result": {
    "resultType": "input_required",
    "inputRequests": {
      "summary": {
        "method": "sampling/createMessage",
        "params": {
          "messages": [{ "role": "user", "content": { "type": "text", "text": "Ниже лог..." } }],
          "systemPrompt": "Ты пишешь краткие технические описания багов на русском языке.",
          "maxTokens": 200
        }
      }
    },
    "requestState": "gAAAAB...<непрозрачно>"
  }
}