"""Tool definitions — Admin-facing actions the KI can plan. Each tool: name, description, JSON Schema for args, and a handler that is only ever called from the `execute` endpoint after the user confirmed the Card. """ from __future__ import annotations from sqlalchemy.orm import Session from core.events import event_bus from core.settings_service import set_setting from apps.ai_core.tools import ToolSpec, register_tool from apps.catalog.models import Category, Product from apps.catalog.projector import project_category, project_product # ---- settings.update ------------------------------------------------ def _handler_settings_update(args: dict, db: Session) -> dict: key = args["key"] value = args["value"] set_setting(db, key, value) return {"key": key, "value": value} SETTINGS_UPDATE = ToolSpec( name="settings.update", description="Update a shop-wide setting (e.g. shop name, currency, support email).", args_schema={ "type": "object", "required": ["key", "value"], "properties": { "key": { "type": "string", "description": "Setting key, e.g. 'core.shop_name'.", }, "value": { "description": "New value (string / number / boolean).", }, }, }, handler=_handler_settings_update, examples=[ {"key": "core.shop_name", "value": "TEST123"}, {"key": "core.support_email", "value": "help@example.com"}, ], ) # ---- catalog.product.create ---------------------------------------- def _coalesce(value, default): """Return default if value is None or missing; otherwise the value.""" return default if value is None else value def _handler_product_create(args: dict, db: Session) -> dict: if db.query(Product).filter_by(sku=args["sku"]).first(): raise ValueError(f"SKU already exists: {args['sku']}") name_de = _coalesce(args.get("name_de"), "") name_en = _coalesce(args.get("name_en"), name_de) desc_de = _coalesce(args.get("description_de"), "") desc_en = _coalesce(args.get("description_en"), desc_de) category_id = args.get("category_id") if category_id in ("", 0): category_id = None p = Product( sku=args["sku"], name={"de": name_de, "en": name_en}, description={"de": desc_de, "en": desc_en}, price=float(args["price"]), currency=_coalesce(args.get("currency"), "EUR") or "EUR", stock=int(_coalesce(args.get("stock"), 0) or 0), active=bool(_coalesce(args.get("active"), True)), image_url=_coalesce(args.get("image_url"), "") or "", category_id=category_id, attributes=_coalesce(args.get("attributes"), {}) or {}, ) db.add(p) db.commit() db.refresh(p) project_product(db, p.id) event_bus.publish("product.created", {"id": p.id, "sku": p.sku}, db=db) return {"id": p.id, "sku": p.sku} PRODUCT_CREATE = ToolSpec( name="catalog.product.create", description="Create a new product in the catalog.", args_schema={ "type": "object", "required": ["sku", "name_de", "price"], "properties": { "sku": {"type": "string"}, "name_de": {"type": "string"}, "name_en": {"type": "string"}, "description_de": {"type": "string"}, "description_en": {"type": "string"}, "price": {"type": "number"}, "currency": {"type": "string", "default": "EUR"}, "stock": {"type": "integer", "default": 0}, "active": {"type": "boolean", "default": True}, "image_url": {"type": "string"}, "category_id": {"type": "integer"}, "attributes": {"type": "object"}, }, }, handler=_handler_product_create, examples=[ { "sku": "TS-GREEN-M", "name_de": "Grünes T-Shirt", "name_en": "Green T-Shirt", "price": 19.90, "stock": 42, "attributes": {"color": "green", "size": "M"}, } ], ) # ---- catalog.product.update ---------------------------------------- def _handler_product_update(args: dict, db: Session) -> dict: pid = int(args["id"]) p = db.get(Product, pid) if not p: raise ValueError(f"Product {pid} not found") if "name_de" in args or "name_en" in args: p.name = { "de": args.get("name_de", p.name.get("de", "")), "en": args.get("name_en", p.name.get("en", "")), } if "description_de" in args or "description_en" in args: p.description = { "de": args.get("description_de", p.description.get("de", "")), "en": args.get("description_en", p.description.get("en", "")), } for f in ("price", "currency", "stock", "active", "image_url", "category_id"): if f in args: setattr(p, f, args[f]) if "attributes" in args: p.attributes = args["attributes"] db.commit() db.refresh(p) project_product(db, p.id) event_bus.publish("product.updated", {"id": p.id}, db=db) return {"id": p.id, "sku": p.sku} PRODUCT_UPDATE = ToolSpec( name="catalog.product.update", description="Update fields of an existing product.", args_schema={ "type": "object", "required": ["id"], "properties": { "id": {"type": "integer"}, "name_de": {"type": "string"}, "name_en": {"type": "string"}, "description_de": {"type": "string"}, "description_en": {"type": "string"}, "price": {"type": "number"}, "stock": {"type": "integer"}, "active": {"type": "boolean"}, "image_url": {"type": "string"}, "category_id": {"type": "integer"}, "attributes": {"type": "object"}, }, }, handler=_handler_product_update, examples=[{"id": 5, "price": 24.90, "stock": 10}], ) # ---- catalog.category.create -------------------------------------- def _handler_category_create(args: dict, db: Session) -> dict: if db.query(Category).filter_by(slug=args["slug"]).first(): raise ValueError(f"Slug exists: {args['slug']}") c = Category( slug=args["slug"], name={"de": args.get("name_de", ""), "en": args.get("name_en", args.get("name_de", ""))}, parent_id=args.get("parent_id"), sort_order=int(args.get("sort_order", 0)), ) db.add(c) db.commit() db.refresh(c) project_category(db, c.id) event_bus.publish("category.created", {"id": c.id}, db=db) return {"id": c.id, "slug": c.slug} CATEGORY_CREATE = ToolSpec( name="catalog.category.create", description="Create a new category.", args_schema={ "type": "object", "required": ["slug", "name_de"], "properties": { "slug": {"type": "string"}, "name_de": {"type": "string"}, "name_en": {"type": "string"}, "parent_id": {"type": "integer"}, "sort_order": {"type": "integer"}, }, }, handler=_handler_category_create, examples=[{"slug": "accessoires", "name_de": "Accessoires", "name_en": "Accessories"}], ) # ---- registration ------------------------------------------------- ALL_TOOLS = [SETTINGS_UPDATE, PRODUCT_CREATE, PRODUCT_UPDATE, CATEGORY_CREATE] def register_all() -> None: for t in ALL_TOOLS: register_tool(t)