diff --git a/ex_app/lib/all_tools/slide_deck.py b/ex_app/lib/all_tools/slide_deck.py new file mode 100644 index 0000000..0129ab0 --- /dev/null +++ b/ex_app/lib/all_tools/slide_deck.py @@ -0,0 +1,62 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Slide deck generation through the marp_slides ExApp.""" +from langchain_core.tools import tool +from nc_py_api import AsyncNextcloudApp + +from ex_app.lib.all_tools.lib.impulse import ImpulseRadius, impulse +from ex_app.lib.all_tools.lib.task_processing import run_task + +TASK_TYPE_ID = "marp_slides:slide_deck_generation" + + +async def get_tools(nc: AsyncNextcloudApp): + + @tool + @impulse(ImpulseRadius.SELF) + async def create_slide_deck(prompt: str, title: str = "Slide deck") -> str: + """Create a presentation as a PPTX slide deck from a deck brief. + + The deck itself is written by the presentation tool's own AI, so pass a + brief and not slides, and no Marp or Markdown. Expand the user's request + into a self-contained brief: the topic and what it should cover, the + audience, the desired number of slides when the user indicated one, the + tone, and the language the user writes in, which is also the language + the deck and your reply should be in. Derive a short file title that + names the deck's subject instead of a generic one. + :param prompt: the deck brief: topic, audience, desired slide count if known, tone, language + :param title: a good file title for the generated deck, derived from the user's request + :return: a download link to the generated presentation + """ + url = ( + await nc.ocs( + "GET", + "/ocs/v2.php/apps/app_api/api/v1/info/nextcloud_url/absolute", + json={"url": "ocs/v2.php/apps/assistant/api/v1/task"}, + ) + )["absolute_url"] + + # The task input shape has no title slot, so the title rides at the top + # of the brief: the provider's AI names the deck and its file from it. + task = await run_task( + nc, + TASK_TYPE_ID, + { + "prompt": f"Deck title: {title}\n\n{prompt}", + "output_format": "pptx", + }, + ) + return f"{url}/{task.id}/output-file/{task.output['slide_deck']}/download" + + return [ + create_slide_deck, + ] + + +def get_category_name(): + return "Slide deck generation" + + +async def is_available(nc: AsyncNextcloudApp): + tasktypes = (await nc.ocs("GET", "/ocs/v2.php/taskprocessing/tasktypes"))["types"].keys() + return TASK_TYPE_ID in tasktypes diff --git a/tests/test_slide_deck_tool.py b/tests/test_slide_deck_tool.py new file mode 100644 index 0000000..ce5ba59 --- /dev/null +++ b/tests/test_slide_deck_tool.py @@ -0,0 +1,109 @@ +# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Hermetic tests for the slide_deck tool module. + +Everything runs against a fake AsyncNextcloudApp that records ocs() calls and +serves canned task-processing responses: no network, no Nextcloud server, and +no environment outside this worktree. The repo-wide ruff config selects S101 +(no assert statements) without a test file exemption, and the shared config +must not be edited for this tool, so the checks below raise AssertionError +from small helpers instead of using the assert keyword. +""" +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from ex_app.lib.all_tools import slide_deck # noqa: E402 + + +class FakeNextcloudApp: + """Records ocs() calls and answers the endpoints the module touches.""" + + def __init__(self, tasktypes=None, nextcloud_url="https://nc.example.test", task_id=42, file_id=1337): + self.tasktypes = dict(tasktypes or {}) + self.nextcloud_url = nextcloud_url + self.task_id = task_id + self.file_id = file_id + self.ocs_calls = [] + + async def ocs(self, method, endpoint, json=None, **_kwargs): + self.ocs_calls.append((method, endpoint, json)) + if endpoint == "/ocs/v2.php/taskprocessing/tasktypes": + return {"types": self.tasktypes} + if endpoint == "/ocs/v2.php/apps/app_api/api/v1/info/nextcloud_url/absolute": + # real endpoint returns a dict (app_api OCSApiController), not a bare string + return {"absolute_url": self.nextcloud_url} + if endpoint == "/ocs/v1.php/taskprocessing/schedule": + return { + "task": { + "id": self.task_id, + "status": "STATUS_SUCCESSFUL", + "output": {"slide_deck": self.file_id}, + } + } + raise AssertionError(f"unexpected ocs call: {method} {endpoint}") + + async def log(self, level, content): + """Sink for run_task's debug logging.""" + + def schedule_calls(self): + return [call for call in self.ocs_calls if call[1] == "/ocs/v1.php/taskprocessing/schedule"] + + +def expect_equal(actual, expected, label): + if actual != expected: + raise AssertionError(f"{label}: expected {expected!r}, got {actual!r}") + + +async def call_slide_deck(nc, **tool_args): + tools = await slide_deck.get_tools(nc) + expect_equal([t.name for t in tools], ["create_slide_deck"], "tool names") + return await tools[0].ainvoke(tool_args) + + +def test_schedules_brief_task_with_marp_task_type(): + nc = FakeNextcloudApp() + brief = "A deck about tending bonsai for beginners, 8 slides, calm tone, German" + asyncio.run(call_slide_deck(nc, prompt=brief, title="Bonsai basics")) + + expect_equal(len(nc.schedule_calls()), 1, "number of schedule calls") + (schedule,) = nc.schedule_calls() + method, endpoint, payload = schedule + expect_equal(method, "POST", "schedule method") + expect_equal(endpoint, "/ocs/v1.php/taskprocessing/schedule", "schedule endpoint") + expect_equal(payload["type"], "marp_slides:slide_deck_generation", "task type id") + expect_equal(payload["appId"], "context_agent", "scheduling app id") + expect_equal( + payload["input"], + { + "prompt": f"Deck title: Bonsai basics\n\n{brief}", + "output_format": "pptx", + }, + "task input payload", + ) + + +def test_builds_download_link_from_slide_deck_output(): + nc = FakeNextcloudApp(task_id=57, file_id=909) + link = asyncio.run(call_slide_deck(nc, prompt="Topic", title="T")) + expect_equal(link, "https://nc.example.test/57/output-file/909/download", "download link") + + +def test_default_title_rides_in_the_brief(): + nc = FakeNextcloudApp() + asyncio.run(call_slide_deck(nc, prompt="Topic")) + ((_, _, payload),) = nc.schedule_calls() + expect_equal(payload["input"]["prompt"].splitlines()[0], "Deck title: Slide deck", "default title line") + expect_equal(payload["input"]["output_format"], "pptx", "output format with default title") + + +def test_is_available_true_when_task_type_registered(): + nc = FakeNextcloudApp(tasktypes={"marp_slides:slide_deck_generation": {}, "core:text2text": {}}) + expect_equal(asyncio.run(slide_deck.is_available(nc)), True, "availability with the ExApp registered") + + +def test_is_available_false_without_task_type(): + nc = FakeNextcloudApp(tasktypes={"core:text2text": {}}) + expect_equal(asyncio.run(slide_deck.is_available(nc)), False, "availability without the ExApp")