"""Инструменты MCP: управление Synapse ИИ-агентом.
Каждый тул — синхронная функция (FastMCP гоняет её в threadpool — совместимо
с синхронной SQLAlchemy-сессией). Логика записи НЕ дублируется: тулы
переиспользуют admin-route функции (admin_routes.py) напрямую с
user=MCP_USER, db=<сессия> — Depends не срабатывают, та же валидация и те же
коды ошибок. HTTPException -> {"error": status, "detail"}.
"""
import uuid
from typing import Any
from fastapi import HTTPException
from pydantic import BaseModel, ValidationError
from sqlalchemy import func, select, text
from app.api import admin_routes
from app.api.admin_schemas import (
KeyIn,
RuleIn,
RulePatch,
SettingsPut,
SourceIn,
TargetIn,
TargetPatch,
TypeIn,
)
from app.auth.deps import AuthenticatedUser
from app.config import get_settings
from app.database import SessionLocal
from app.models import (
ChannelTarget,
Delivery,
Event,
NotificationType,
PushSubscription,
RoutingRule,
Source,
)
from app.worker.celery_app import celery_app
# Синтетический superadmin: require_admin пропускает, автор — MCP-токен.
MCP_USER = AuthenticatedUser(
user_id="mcp", email="mcp@synapse.local", email_verified=True, system_role="superadmin"
)
LIMIT_CAP = 500
def _run(fn) -> Any:
"""Сессия на вызов тулза; HTTPException/pydantic ValidationError -> dict ошибки."""
try:
with SessionLocal() as db:
return fn(db)
except HTTPException as err:
return {"error": err.status_code, "detail": str(err.detail)}
except ValidationError as err:
return {"error": 422, "detail": err.errors()[0].get("msg", "невалидно")}
def _patch(model_cls: type[BaseModel], **kwargs) -> BaseModel:
"""PATCH-схема только из переданных полей: None = «не менять»
(exclude_unset в route-функции видит незаданные поля незаданными)."""
return model_cls.model_validate({k: v for k, v in kwargs.items() if v is not None})
def register_tools(mcp) -> None:
@mcp.tool()
def system_status() -> dict:
"""Здоровье Synapse: БД, Celery-воркер, конфигурация (SSO)."""
def fn(db):
db.execute(text("SELECT 1"))
return {"database": "ok"}
out = _run(fn)
if "error" in out:
return {"database": "unavailable", **out}
try:
pong = celery_app.send_task("synapse.ping").get(timeout=5)
out.update(worker=pong, sso_configured=get_settings().gauth_configured())
except Exception: # noqa: BLE001 — недоступный воркер не ломает статус
out.update(worker="unavailable", sso_configured=get_settings().gauth_configured())
return out
@mcp.tool()
def stats_get() -> dict:
"""Сводка: события и доставки по статусам, число живых
sources/types/targets/rules, число deliveries в ожидании."""
def fn(db):
ev = dict(
db.execute(select(Event.status, func.count(Event.id)).group_by(Event.status)).all()
)
dv = dict(
db.execute(
select(Delivery.status, func.count(Delivery.id)).group_by(Delivery.status)
).all()
)
return {
"events": ev,
"deliveries": dv,
"pending_deliveries": dv.get("pending", 0),
"sources": db.scalar(
select(func.count(Source.id)).where(Source.deleted_at.is_(None))
),
"types": db.scalar(
select(func.count(NotificationType.id)).where(
NotificationType.deleted_at.is_(None)
)
),
"targets": db.scalar(
select(func.count(ChannelTarget.id)).where(
ChannelTarget.deleted_at.is_(None)
)
),
"rules": db.scalar(
select(func.count(RoutingRule.id)).where(RoutingRule.deleted_at.is_(None))
),
}
return _run(fn)
# --- источники и ключи (выдача ключей клиентам, docs/07) ---
@mcp.tool()
def sources_list(
query: str | None = None, include_archived: bool = False, limit: int = 100
) -> list[dict]:
"""Источники событий. query — подстрока в name/label/description;
include_archived=true — показать и заархивированные (deleted_at != null)."""
def fn(db):
rows = admin_routes.list_sources(
include_archived=include_archived, user=MCP_USER, db=db
)
out = [r.model_dump(mode="json") for r in rows]
if query:
ql = query.lower()
out = [
r for r in out
if ql in r["name"].lower()
or ql in (r.get("label") or "").lower()
or ql in (r.get("description") or "").lower()
]
return out[: max(1, min(limit, LIMIT_CAP))]
return _run(fn)
@mcp.tool()
def source_create(name: str, label: str | None = None, description: str | None = None) -> dict:
"""Зарегистрировать источник событий (сервис). name — slug [a-z0-9._-],
label/description — человекочитаемое «что это за сервис»."""
def fn(db):
src = admin_routes.create_source(
SourceIn(name=name, label=label, description=description),
user=MCP_USER, db=db,
)
return src.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def source_archive(source_id: int) -> dict:
"""Заархивировать источник (это не удаление): ключ перестаёт принимать
события (401), source_restore возвращает в строй."""
def fn(db):
admin_routes.delete_source(source_id, user=MCP_USER, db=db)
return {"ok": True, "archived": source_id}
return _run(fn)
@mcp.tool()
def source_restore(source_id: int) -> dict:
"""Вернуть заархивированный источник в строй. Конфликт имени → 409."""
def fn(db):
src = admin_routes.restore_source(source_id, user=MCP_USER, db=db)
return src.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def keys_list(source_id: int) -> list[dict]:
"""API-ключи источника: хэш и хвост — plaintext никогда не возвращается,
полный токен виден один раз в key_issue."""
def fn(db):
keys = admin_routes.list_keys(source_id, user=MCP_USER, db=db)
return [k.model_dump(mode="json") for k in keys]
return _run(fn)
@mcp.tool()
def key_issue(source_id: int, name: str) -> dict:
"""Выдать API-ключ клиенту (сервису-источнику). Полный токен `syn_...`
в поле token возвращается ОДИН РАЗ — передать клиенту (в его .env);
в Synapse остаётся только хэш. Повторный вызов = новый токен."""
def fn(db):
created = admin_routes.create_key(
source_id, KeyIn(name=name), user=MCP_USER, db=db
)
return created.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def key_revoke(source_id: int, key_id: int) -> dict:
"""Отозвать API-ключ источника (события с ним начнут получать 401)."""
def fn(db):
key = admin_routes.revoke_key(source_id, key_id, user=MCP_USER, db=db)
return key.model_dump(mode="json")
return _run(fn)
# --- типы уведомлений (тройка source/subject/action) ---
@mcp.tool()
def types_list(
source_id: int | None = None,
query: str | None = None,
include_archived: bool = False,
limit: int = 200,
) -> list[dict]:
"""Зарегистрированные типы уведомлений (тройка source/subject/action).
Событие с незарегистрированной тройкой приём отклоняет (422)."""
def fn(db):
rows = admin_routes.list_types(
include_archived=include_archived, user=MCP_USER, db=db
)
out = [r.model_dump(mode="json") for r in rows]
if source_id is not None:
out = [r for r in out if r["source_id"] == source_id]
if query:
ql = query.lower()
out = [
r for r in out
if ql in r["subject"].lower()
or ql in r["action"].lower()
or ql in (r.get("description") or "").lower()
]
return out[: max(1, min(limit, LIMIT_CAP))]
return _run(fn)
@mcp.tool()
def type_register(
source_id: int,
subject: str,
action: str,
description: str | None = None,
payload_schema: dict | None = None,
) -> dict:
"""Зарегистрировать тип уведомления: тройка (source_id, subject, action).
payload_schema — опциональная JSON Schema payload'а (валидация мягкая, в воркере)."""
def fn(db):
nt = admin_routes.create_type(
TypeIn(
source_id=source_id, subject=subject, action=action,
payload_schema=payload_schema, description=description,
),
user=MCP_USER, db=db,
)
return nt.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def type_archive(type_id: int) -> dict:
"""Заархивировать тип: приём его тройки начнёт отвечать
422 «тип не зарегистрирован»; type_restore вернёт в строй."""
def fn(db):
admin_routes.delete_type(type_id, user=MCP_USER, db=db)
return {"ok": True, "archived": type_id}
return _run(fn)
@mcp.tool()
def type_restore(type_id: int) -> dict:
"""Вернуть заархивированный тип в строй (источник должен быть жив)."""
def fn(db):
nt = admin_routes.restore_type(type_id, user=MCP_USER, db=db)
return nt.model_dump(mode="json")
return _run(fn)
# --- цели каналов ---
@mcp.tool()
def targets_list(
channel: str | None = None,
enabled: bool | None = None,
include_archived: bool = False,
limit: int = 200,
) -> list[dict]:
"""Цели каналов (TG-чат, email-адрес, s2s-точка). channel — фильтр."""
def fn(db):
rows = admin_routes.list_targets(
include_archived=include_archived, user=MCP_USER, db=db
)
out = [r.model_dump(mode="json") for r in rows]
if channel:
out = [r for r in out if r["channel"] == channel]
if enabled is not None:
out = [r for r in out if r["enabled"] == enabled]
return out[: max(1, min(limit, LIMIT_CAP))]
return _run(fn)
@mcp.tool()
def target_create(
channel: str, name: str, config: dict, description: str | None = None
) -> dict:
"""Создать цель канала. channel ∈ telegram|email|s2s|internal_log;
config — идентификатор цели (s2s: {"endpoint": …, "token_ref": "navi-rei"}).
Секреты целей — в .env/gnexus-creds, в config только token_ref."""
def fn(db):
t = admin_routes.create_target(
TargetIn(channel=channel, name=name, config=config, description=description),
user=MCP_USER, db=db,
)
return t.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def target_patch(
target_id: int, name: str | None = None, description: str | None = None,
config: dict | None = None, enabled: bool | None = None,
) -> dict:
"""Частичное обновление цели (переданное — меняется, None — «не менять»).
Канал не меняется (ссылки из правил)."""
def fn(db):
t = admin_routes.patch_target(
target_id,
_patch(TargetPatch, name=name, description=description,
config=config, enabled=enabled),
user=MCP_USER, db=db,
)
return t.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def target_archive(target_id: int) -> dict:
"""Заархивировать цель: правила продолжат матчить, но доставки в неё
запишутся skipped «цель в архиве»; restore вернёт в строй."""
def fn(db):
admin_routes.delete_target(target_id, user=MCP_USER, db=db)
return {"ok": True, "archived": target_id}
return _run(fn)
@mcp.tool()
def target_restore(target_id: int) -> dict:
"""Вернуть заархивированную цель в строй. Конфликт имени → 409."""
def fn(db):
t = admin_routes.restore_target(target_id, user=MCP_USER, db=db)
return t.model_dump(mode="json")
return _run(fn)
# --- правила маршрутизации ---
@mcp.tool()
def rules_list(
enabled: bool | None = None, include_archived: bool = False, limit: int = 200
) -> list[dict]:
"""Правила маршрутизации (условия -> действия). weight — порядок в режиме first."""
def fn(db):
rows = admin_routes.list_rules(include_archived=include_archived, user=MCP_USER, db=db)
out = [r.model_dump(mode="json") for r in rows]
if enabled is not None:
out = [r for r in out if r["enabled"] == enabled]
return out[: max(1, min(limit, LIMIT_CAP))]
return _run(fn)
@mcp.tool()
def rule_create(
name: str,
actions: list[dict],
conditions: dict | None = None,
template: str | None = None,
throttle_seconds: int | None = None,
weight: int = 0,
) -> dict:
"""Создать правило: действия [{"channel": "telegram|email|s2s|internal_log|user|push",
"target_id": id|null, "template": "{{ payload.x }}"}]; условия
{"source": "имя"|null, "subjects": [], "actions": [], "priority_min": "normal",
"payload": {...}}. Каналы user/push целей не имеют (адресат payload.user_id)."""
def fn(db):
rule = admin_routes.create_rule(
RuleIn.model_validate({
"name": name, "actions": actions,
"conditions": conditions or {}, "template": template,
"throttle_seconds": throttle_seconds, "weight": weight,
}),
user=MCP_USER, db=db,
)
return rule.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def rule_patch(
rule_id: int, name: str | None = None, enabled: bool | None = None,
conditions: dict | None = None, template: str | None = None,
throttle_seconds: int | None = None, weight: int | None = None,
actions: list[dict] | None = None,
) -> dict:
"""Частично обновить правило (вкл/выкл, условия, действия, шаблон, вес)."""
def fn(db):
rule = admin_routes.patch_rule(
rule_id,
_patch(RulePatch, name=name, enabled=enabled, conditions=conditions,
template=template, throttle_seconds=throttle_seconds,
weight=weight, actions=actions),
user=MCP_USER, db=db,
)
return rule.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def rule_archive(rule_id: int) -> dict:
"""Заархивировать правило (действия сохраняются — restore вернёт целиком)."""
def fn(db):
admin_routes.delete_rule(rule_id, user=MCP_USER, db=db)
return {"ok": True, "archived": rule_id}
return _run(fn)
@mcp.tool()
def rule_restore(rule_id: int) -> dict:
"""Вернуть заархивированное правило в строй."""
def fn(db):
rule = admin_routes.restore_rule(rule_id, user=MCP_USER, db=db)
return rule.model_dump(mode="json")
return _run(fn)
# --- поток: события и доставки ---
@mcp.tool()
def events_list(
limit: int = 50, status: str | None = None, source_name: str | None = None
) -> list[dict]:
"""Последние события (конверт + payload + доставки). status ∈
queued|processing|done|failed|expired; source_name — фильтр по имени источника."""
def fn(db):
source_id = None
if source_name:
src = db.execute(
select(Source.id).where(Source.name == source_name)
).scalar_one_or_none()
source_id = src if src is not None else -1 # нет источника — пусто
rows = admin_routes.list_events(
user=MCP_USER, db=db, limit=limit, status_filter=status, source_id=source_id
)
return [r.model_dump(mode="json") for r in rows]
return _run(fn)
@mcp.tool()
def event_get(event_id: str) -> dict:
"""Событие по id: конверт, статусы и доставки в каналы."""
def fn(db):
ev = admin_routes.get_event(uuid.UUID(event_id), user=MCP_USER, db=db)
return ev.model_dump(mode="json")
return _run(fn)
@mcp.tool()
def deliveries_list(
limit: int = 100, status: str | None = None, channel: str | None = None
) -> list[dict]:
"""Журнал доставок: канал, цель, статус, попытки, ошибки ретраев.
status ∈ pending|delivered|skipped|failed."""
def fn(db):
rows = admin_routes.list_deliveries(
user=MCP_USER, db=db, limit=limit, status_filter=status, channel=channel
)
return [r.model_dump(mode="json") for r in rows]
return _run(fn)
@mcp.tool()
def push_subscriptions_list(user_id: str | None = None, limit: int = 100) -> list[dict]:
"""Web-push подписки пользователей (адресат — payload.user_id).
user_id — фильтр по sub gnexus-auth."""
def fn(db):
q = select(PushSubscription).order_by(PushSubscription.id.desc())
if user_id:
q = q.where(PushSubscription.user_id == user_id)
rows = db.execute(q.limit(max(1, min(limit, LIMIT_CAP)))).scalars().all()
return [
{
"id": s.id, "user_id": s.user_id, "endpoint": s.endpoint,
"ua": s.ua, "created_at": s.created_at.isoformat(),
}
for s in rows
]
return _run(fn)
@mcp.tool()
def push_subscription_delete(subscription_id: int) -> dict:
"""Удалить web-push подписку: единственное ЖЁСТКОЕ удаление в Synapse —
устройство пользователя больше не подписано."""
def fn(db):
sub = db.get(PushSubscription, subscription_id)
if sub is None:
raise HTTPException(status_code=404, detail="Подписка не найдена")
db.delete(sub)
db.commit()
return {"ok": True, "deleted": subscription_id}
return _run(fn)
# --- настройки ---
@mcp.tool()
def settings_get() -> list[dict]:
"""Реестр настроек Synapse: дефолт .env + оверрайды админки.
Секреты write-only — value всегда null, факт наличия в set."""
def fn(db):
rows = admin_routes.list_settings(user=MCP_USER, db=db)
return [r.model_dump(mode="json") for r in rows]
return _run(fn)
@mcp.tool()
def settings_put(values: dict) -> dict:
"""Установить оверрайды настроек (поверх .env). Правила: строка — set,
null — вернуть дефолт .env, "" у секрета — «не менять». Нарушение — 422."""
def fn(db):
return admin_routes.put_settings(
SettingsPut.model_validate({"values": values}), user=MCP_USER, db=db
)
return _run(fn)
# --- приём контрольного события ---
@mcp.tool()
def send_test_event(
source_name: str, subject: str, action: str, payload: dict | None = None,
priority: str = "normal", ttl_seconds: int | None = None,
) -> dict:
"""Контрольное событие «как бы от источника» (source_name): полный
маршрут в воркере -> доставки по правилам. Для проверки нового правила.
Тройка должна быть зарегистрирована (type_register)."""
from app.api.events_routes import accept_event
from app.api.schemas import EventEnvelope
def fn(db):
src = db.execute(
select(Source).where(Source.name == source_name)
).scalar_one_or_none()
if src is None:
raise HTTPException(status_code=404, detail=f"Источник '{source_name}' не найден")
if src.deleted_at is not None:
raise HTTPException(
status_code=422, detail=f"Источник '{source_name}' в архиве — сначала restore"
)
envelope = EventEnvelope(
source=source_name, subject=subject, action=action, priority=priority,
payload=payload or {}, ttl_seconds=ttl_seconds,
)
event = accept_event(db, src, envelope)
db.commit()
celery_app.send_task("synapse.ingest", args=[str(event.id)])
return {"id": str(event.id), "status": event.status, "source": source_name}
return _run(fn)
