fix(mcp): batch_wait и refresh асинхронные, refresh по умолчанию без ожидания (R-12, I-3)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev
parent
4f055a80d0
commit
1bc6ebe534
@ -62,10 +62,10 @@ uv run jira-hub import ../admin/jira-sync
|
||||
| `notes_set` | `key`, поля заметок | обновлённые заметки |
|
||||
| `batch_create` | `title`, `commands` | `id`, оценка длительности, число команд |
|
||||
| `batch_status` | `id` | статусы команд, ошибки, оценка остатка |
|
||||
| `batch_wait` | `id`, `timeout_minutes` (до 60) | как `batch_status`, по завершении или таймауту |
|
||||
| `batch_wait` | `id`, `timeout_minutes` (до 60) | как `batch_status`, по завершении или таймауту; ожидание асинхронное, сервер не блокируется |
|
||||
| `batches` | `limit` | последние пакеты |
|
||||
| `command_retry`, `batch_cancel` | `id` | новый пакет / результат отмены |
|
||||
| `refresh` | `keys` или `full`, `wait` | `sync_run` или id пакета |
|
||||
| `refresh` | `keys` или `full`, `wait` (по умолчанию `false`: только ставит пакет; `true` — ждёт до 30 мин) | id пакета или итог пакета |
|
||||
| `agent_status` | нет | онлайн/офлайн, версия, текущая команда, 429 за сутки |
|
||||
| `sync_log` | `limit` | последние загрузки состояния |
|
||||
| `settings_get`, `settings_set` | ключ, значение | значение / новая `settings_version` |
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
"""MCP-сервер хаба (stdio, FastMCP) — инструменты сессии Claude поверх HTTP API (specs/mcp/spec.md).
|
||||
Автор команд из MCP — «claude». Если хаб выключен, инструмент возвращает подсказку, а не исключение."""
|
||||
import asyncio
|
||||
import functools
|
||||
import inspect
|
||||
import json
|
||||
import time
|
||||
|
||||
@ -18,7 +20,23 @@ def _client() -> HubClient:
|
||||
return HubClient.from_home()
|
||||
|
||||
|
||||
def _hub_error(e: HubError) -> str:
|
||||
return f"ошибка хаба {e.status} {e.error}: {e.detail or ''}".rstrip(": ")
|
||||
|
||||
|
||||
def guard(fn):
|
||||
"""Обёртка инструмента: сбои хаба превращаются в текст. Поддерживает и обычные, и async-функции."""
|
||||
if inspect.iscoroutinefunction(fn):
|
||||
@functools.wraps(fn)
|
||||
async def awrapper(*args, **kwargs):
|
||||
try:
|
||||
return await fn(*args, **kwargs)
|
||||
except HubUnavailable:
|
||||
return OFFLINE
|
||||
except HubError as e:
|
||||
return _hub_error(e)
|
||||
return awrapper
|
||||
|
||||
@functools.wraps(fn)
|
||||
def wrapper(*args, **kwargs):
|
||||
try:
|
||||
@ -26,7 +44,7 @@ def guard(fn):
|
||||
except HubUnavailable:
|
||||
return OFFLINE
|
||||
except HubError as e:
|
||||
return f"ошибка хаба {e.status} {e.error}: {e.detail or ''}".rstrip(": ")
|
||||
return _hub_error(e)
|
||||
return wrapper
|
||||
|
||||
|
||||
@ -138,15 +156,15 @@ def batch_status(id: int) -> str:
|
||||
|
||||
@mcp.tool()
|
||||
@guard
|
||||
def batch_wait(id: int, timeout_minutes: int = 10) -> str:
|
||||
"""Ждать завершения пакета (до 60 мин), вернуть итог по командам."""
|
||||
async def batch_wait(id: int, timeout_minutes: int = 10) -> str:
|
||||
"""Ждать завершения пакета (до 60 мин), вернуть итог по командам. Ожидание не блокирует event loop сервера."""
|
||||
deadline = time.monotonic() + min(max(int(timeout_minutes), 1), 60) * 60
|
||||
client = _client()
|
||||
while True:
|
||||
b = client.get(f"/batches/{id}")
|
||||
b = await asyncio.to_thread(client.get, f"/batches/{id}")
|
||||
if b["status"] in ("done", "cancelled") or time.monotonic() >= deadline:
|
||||
return fmt_batch(b) + ("" if b["status"] in ("done", "cancelled") else "\n(таймаут ожидания, пакет ещё выполняется)")
|
||||
time.sleep(POLL_SECONDS)
|
||||
await asyncio.sleep(POLL_SECONDS)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
@ -174,12 +192,13 @@ def batch_cancel(id: int) -> str:
|
||||
|
||||
@mcp.tool()
|
||||
@guard
|
||||
def refresh(keys: list[str] | None = None, full: bool = False, wait: bool = True) -> str:
|
||||
"""Обновить зеркало из Jira: по ключам или полностью (базовая JQL). wait — дождаться результата."""
|
||||
async def refresh(keys: list[str] | None = None, full: bool = False, wait: bool = False) -> str:
|
||||
"""Обновить зеркало из Jira: по ключам или полностью (базовая JQL). По умолчанию только ставит пакет;
|
||||
wait=true — дождаться результата (до 30 мин)."""
|
||||
body = {"full": True} if full or not keys else {"keys": keys}
|
||||
body["author"] = AUTHOR
|
||||
b = _client().post("/refresh", body)
|
||||
return batch_wait(b["id"], 30) if wait else f"пакет #{b['id']} «Обновление» поставлен"
|
||||
b = await asyncio.to_thread(_client().post, "/refresh", body)
|
||||
return await batch_wait(b["id"], 30) if wait else f"пакет #{b['id']} «Обновление» поставлен"
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
|
||||
@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import json
|
||||
import re
|
||||
import threading
|
||||
@ -23,6 +24,7 @@ def live(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(mcp_server, "POLL_SECONDS", 0.05)
|
||||
yield {"agent": HubClient(base, tok["agent"])}
|
||||
srv.shutdown()
|
||||
srv.server_close()
|
||||
|
||||
|
||||
def test_offline_message(monkeypatch):
|
||||
@ -48,13 +50,14 @@ def test_tools_roundtrip(live):
|
||||
live["agent"].post(f"/agent/commands/{cmd['id']}/result", {"status": "failed", "error": "HTTP 400 поле"})
|
||||
tail = live["agent"].post("/agent/claim", {})["command"]
|
||||
live["agent"].post(f"/agent/commands/{tail['id']}/result", {"status": "done"})
|
||||
waited = mcp_server.batch_wait(batch_id, timeout_minutes=1)
|
||||
waited = asyncio.run(mcp_server.batch_wait(batch_id, timeout_minutes=1))
|
||||
assert "done" in waited and "failed" in waited and "HTTP 400" in waited
|
||||
assert f"#{batch_id}" in mcp_server.batches(limit=5)
|
||||
assert "Повтор команды" in mcp_server.command_retry(cmd["id"])
|
||||
assert "cancelled" in mcp_server.batch_cancel(batch_id + 1)
|
||||
r = mcp_server.refresh(full=True, wait=False)
|
||||
r = asyncio.run(mcp_server.refresh(full=True))
|
||||
assert re.fullmatch(r"пакет #\d+ «Обновление» поставлен", r)
|
||||
assert re.fullmatch(r"пакет #\d+ «Обновление» поставлен", asyncio.run(mcp_server.refresh(keys=["A-1"], wait=False)))
|
||||
assert "онлайн" in mcp_server.agent_status()
|
||||
assert "120" in mcp_server.settings_get("pacing")
|
||||
assert "settings_version" in mcp_server.settings_set("pacing", json.dumps({"refresh": 1}))
|
||||
@ -98,10 +101,11 @@ def _fake_time(monkeypatch, step=1.0):
|
||||
clock["now"] += step
|
||||
return clock["now"]
|
||||
|
||||
def sleep(_):
|
||||
async def sleep(_):
|
||||
clock["sleeps"] += 1
|
||||
|
||||
monkeypatch.setattr(mcp_server, "time", SimpleNamespace(monotonic=monotonic, sleep=sleep))
|
||||
monkeypatch.setattr(mcp_server, "time", SimpleNamespace(monotonic=monotonic))
|
||||
monkeypatch.setattr(mcp_server, "asyncio", SimpleNamespace(sleep=sleep, to_thread=asyncio.to_thread))
|
||||
return clock
|
||||
|
||||
|
||||
@ -109,7 +113,7 @@ def test_batch_wait_polls_until_done(monkeypatch):
|
||||
fake = FakeClient(["running", "running", "done"])
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: fake)
|
||||
clock = _fake_time(monkeypatch)
|
||||
out = mcp_server.batch_wait(7, timeout_minutes=10)
|
||||
out = asyncio.run(mcp_server.batch_wait(7, timeout_minutes=10))
|
||||
assert out == mcp_server.fmt_batch(_batch("done")) and "таймаут" not in out
|
||||
assert fake.calls == 3 and clock["sleeps"] == 2
|
||||
|
||||
@ -118,7 +122,7 @@ def test_batch_wait_timeout(monkeypatch):
|
||||
fake = FakeClient(["running"])
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: fake)
|
||||
clock = _fake_time(monkeypatch, step=30.0) # 1 минута = 2 шага монотонных часов
|
||||
out = mcp_server.batch_wait(7, timeout_minutes=1)
|
||||
out = asyncio.run(mcp_server.batch_wait(7, timeout_minutes=1))
|
||||
assert "таймаут ожидания" in out
|
||||
assert 1 < fake.calls < 10 and clock["sleeps"] == fake.calls - 1
|
||||
|
||||
@ -128,7 +132,7 @@ def test_batch_wait_clamps_timeout(monkeypatch, asked, minutes):
|
||||
fake = FakeClient(["running"])
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: fake)
|
||||
clock = _fake_time(monkeypatch, step=60.0) # каждый вызов часов = +1 минута
|
||||
mcp_server.batch_wait(7, timeout_minutes=asked)
|
||||
asyncio.run(mcp_server.batch_wait(7, timeout_minutes=asked))
|
||||
# первый вызов часов (1 мин) задаёт старт; дедлайн = старт + minutes; выход, когда now >= дедлайна
|
||||
assert fake.calls == minutes
|
||||
|
||||
@ -137,7 +141,7 @@ def test_client_created_once_per_wait(monkeypatch):
|
||||
fake, made = FakeClient(["running", "running", "done"]), []
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: made.append(1) or fake)
|
||||
_fake_time(monkeypatch)
|
||||
mcp_server.batch_wait(7)
|
||||
asyncio.run(mcp_server.batch_wait(7))
|
||||
assert len(made) == 1
|
||||
|
||||
|
||||
@ -146,3 +150,37 @@ def test_sync_log_content(monkeypatch):
|
||||
"issues_count": 12, "links_count": 5, "jira_calls": 4, "rate_limited": 1}]
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: FakeClient(["done"], runs))
|
||||
assert mcp_server.sync_log() == "#3 full project = AWPCA: 2026-10-11T10:00 → —, задач 12, связей 5, вызовов 4, 429: 1"
|
||||
|
||||
|
||||
def test_async_tools_are_coroutines_and_registered():
|
||||
assert asyncio.iscoroutinefunction(mcp_server.batch_wait) and asyncio.iscoroutinefunction(mcp_server.refresh)
|
||||
tools = {t.name: t for t in asyncio.run(mcp_server.mcp.list_tools())}
|
||||
assert {"batch_wait", "refresh"} <= set(tools)
|
||||
assert tools["refresh"].inputSchema["properties"]["wait"]["default"] is False
|
||||
|
||||
|
||||
def test_async_guard_offline(monkeypatch):
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: HubClient("http://127.0.0.1:9", "t", timeout=1))
|
||||
assert asyncio.run(mcp_server.batch_wait(1)) == mcp_server.OFFLINE
|
||||
assert asyncio.run(mcp_server.refresh(full=True)) == mcp_server.OFFLINE
|
||||
|
||||
|
||||
def test_batch_wait_does_not_block_event_loop(monkeypatch):
|
||||
"""Пока идёт ожидание, другая задача event loop продолжает выполняться."""
|
||||
fake = FakeClient(["running", "running", "done"])
|
||||
monkeypatch.setattr(mcp_server, "_client", lambda: fake)
|
||||
monkeypatch.setattr(mcp_server, "POLL_SECONDS", 0.01)
|
||||
|
||||
async def scenario():
|
||||
ticks = []
|
||||
|
||||
async def ticker():
|
||||
for _ in range(5):
|
||||
ticks.append(1)
|
||||
await asyncio.sleep(0.005)
|
||||
|
||||
out, _ = await asyncio.gather(mcp_server.batch_wait(7), ticker())
|
||||
return out, ticks
|
||||
|
||||
out, ticks = asyncio.run(scenario())
|
||||
assert "done" in out and len(ticks) == 5
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user