feat: support multimodal chat inputs and outputs

Signed-off-by: Lukas Schaefer <lukas@lschaefer.xyz>

Author: lukasdotcom

ex_app/lib/agent.py

@@ -17,7 +17,13 @@
 from ex_app.lib.signature import verify_signature
 from ex_app.lib.signature import add_signature
 from ex_app.lib.graph import AgentState, get_graph
-from ex_app.lib.nc_model import model
+from ex_app.lib.nc_model import (
+	model,
+	MULTIMODAL_INTERACTION,
+	extract_text_content,
+	extract_file_ids,
+	build_multimodal_content,
+)
 from ex_app.lib.tools import get_tools
 from ex_app.lib.memorysaver import MemorySaver
 from ex_app.lib.jsonplus import JsonPlusSerializer
@@ -102,9 +108,11 @@ async def react(
 		nc: AsyncNextcloudApp,
 		stream_output: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
 ):
-	safe_tools, dangerous_tools = await get_tools(nc)
-
+	multimodal = task.get('type') == MULTIMODAL_INTERACTION
 	model.bind_nextcloud(nc)
+	model.multimodal = multimodal
+
+	safe_tools, dangerous_tools = await get_tools(nc)
 
 	tools = dangerous_tools + safe_tools
 
@@ -188,7 +196,13 @@ async def call_model(
 		else:
 			new_input = None
 	else:
-		new_input = {"messages": [("user", task['input']['input'])]}
+		input_attachments = task['input'].get('input_attachments') or []
+		user_text = task['input']['input']
+		if multimodal and input_attachments:
+			user_content = build_multimodal_content(user_text, [int(file_id) for file_id in input_attachments])
+			new_input = {"messages": [HumanMessage(content=user_content)]}
+		else:
+			new_input = {"messages": [("user", user_text)]}
 
 	snapshot_messages = state_snapshot.values.get('messages', [])
 	last_message: AIMessage = AIMessage("")
@@ -252,9 +266,12 @@ async def report_stream_state(force: bool = False):
 	if state_snapshot.next == ('dangerous_tools', ):
 		actions = json.dumps(last_message.tool_calls)
 
-	return {
-		'output': last_message.content,
+	result = {
+		'output': extract_text_content(last_message.content),
 		'actions': actions,
 		'conversation_token': export_conversation(checkpointer),
 		'sources': source_list,
 	}
+	if multimodal:
+		result['output_attachments'] = extract_file_ids(last_message.content)
+	return result

ex_app/lib/main.py

@@ -24,9 +24,13 @@
 from ex_app.lib.agent import react
 from ex_app.lib.logger import log
 from ex_app.lib.mcp_server import UserAuthMiddleware, ToolListMiddleware
-from ex_app.lib.provider import provider
+from ex_app.lib.provider import provider, multimodal_provider
 from ex_app.lib.tools import get_categories
 
+PROVIDERS = [provider, multimodal_provider]
+PROVIDER_IDS = [p.id for p in PROVIDERS]
+TASK_TYPES = [p.task_type for p in PROVIDERS]
+
 from contextvars import ContextVar
 from gettext import translation
 from fastmcp import FastMCP
@@ -113,7 +117,8 @@ async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str:
     # NOTE: `user` is unavailable on this step, so all NC API calls that require it will fail as unauthorized.
     await log(nc, LogLvl.INFO, f"enabled={enabled}")
     if enabled:
-        await nc.providers.task_processing.register(provider)
+        for p in PROVIDERS:
+            await nc.providers.task_processing.register(p)
         app_enabled.set()
         await log(nc, LogLvl.WARNING, f"App enabled: {nc.app_cfg.app_name}")
 
@@ -125,7 +130,8 @@ async def enabled_handler(enabled: bool, nc: AsyncNextcloudApp) -> str:
         await nc.appconfig_ex.set_value('tool_status', json.dumps(pref_settings))
 
     else:
-        await nc.providers.task_processing.unregister(provider.id)
+        for p in PROVIDERS:
+            await nc.providers.task_processing.unregister(p.id)
         app_enabled.clear()
         await log(nc, LogLvl.WARNING, f"App disabled: {nc.app_cfg.app_name}")
     # In case of an error, a non-empty short string should be returned, which will be shown to the NC administrator.
@@ -142,7 +148,7 @@ async def background_thread_task():
                 continue
 
             try:
-                response = await nc.providers.task_processing.next_task([provider.id], [provider.task_type])
+                response = await nc.providers.task_processing.next_task(PROVIDER_IDS, TASK_TYPES)
                 if not response or not 'task' in response:
                     async with NUM_RUNNING_TASKS_LOCK:
                         no_tasks_running = NUM_RUNNING_TASKS == 0
@@ -162,7 +168,14 @@ async def background_thread_task():
             task = response["task"]
             await log(nc, LogLvl.INFO, 'New Task incoming')
             await log(nc, LogLvl.DEBUG, str(task))
-            await log(nc, LogLvl.INFO, str({'input': task['input']['input'], 'confirmation': task['input']['confirmation'], 'conversation_token': '<skipped>', 'memories': task['input'].get('memories', None)}))
+            await log(nc, LogLvl.INFO, str({
+                'type': task.get('type'),
+                'input': task['input']['input'],
+                'confirmation': task['input']['confirmation'],
+                'conversation_token': '<skipped>',
+                'memories': task['input'].get('memories', None),
+                'input_attachments': task['input'].get('input_attachments', None),
+            }))
             tg.create_task(handle_task(task, nc))
 
 NUM_RUNNING_TASKS_LOCK = asyncio.Lock()

ex_app/lib/nc_model.py

@@ -24,6 +24,50 @@
 from ex_app.lib.logger import log
 
 
+TEXT_CHAT_WITH_TOOLS = "core:text2text:chatwithtools"
+MULTIMODAL_CHAT_WITH_TOOLS = "core:text2text:multimodal-chatwithtools"
+MULTIMODAL_INTERACTION = "core:contextagent:multimodal-interaction"
+
+
+def extract_text_content(content: Any) -> str:
+	"""Extract plain text from a string or multimodal content-part list."""
+	if isinstance(content, str):
+		return content
+	if isinstance(content, list):
+		parts = []
+		for part in content:
+			if isinstance(part, dict) and part.get("type") == "text":
+				parts.append(part.get("text", ""))
+			elif isinstance(part, str):
+				parts.append(part)
+		return "".join(parts)
+	return ""
+
+
+def extract_file_ids(content: Any) -> list[int]:
+	"""Extract file IDs from a multimodal content-part list."""
+	if not isinstance(content, list):
+		return []
+	file_ids = []
+	for part in content:
+		if isinstance(part, dict) and part.get("type") == "file" and "file_id" in part:
+			file_ids.append(int(part["file_id"]))
+	return file_ids
+
+
+def build_multimodal_content(text: str, file_ids: list[int] | None = None, task_id: int | None = None) -> list[dict[str, Any]] | str:
+	"""Build multimodal content parts, or plain text when there are no files."""
+	if not file_ids:
+		return text
+	content: list[dict[str, Any]] = [
+		({"type": "file", "file_id": file_id} if task_id is None else {"type": "file", "file_id": file_id, "ocp_task_id": task_id})
+		for file_id in file_ids
+	]
+	if text:
+		content.append({"type": "text", "text": text})
+	return content
+
+
 class Task(BaseModel):
 	id: int
 	status: str
@@ -45,6 +89,7 @@ class ChatWithNextcloud(BaseChatModel):
 	TOOL_OUTPUT_MAX_LENGTH: int = 2000
 	POLL_WAIT_TIME: int = 5
 	STREAMING_POLL_WAIT_TIME: int = 1
+	multimodal: bool = False
 
 	def _generate(self, messages: list[BaseMessage], stop: Optional[list[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any):
 		raise Exception("Use _agenerate instead")
@@ -77,6 +122,8 @@ def _build_task_input(self, messages: list[BaseMessage]) -> dict[str, typing.Any
 		task_input['input'] = ''
 		task_input['tool_message'] = []
 		task_input['tools'] = json.dumps(self.tools)
+		if self.multimodal:
+			task_input['input_attachments'] = []
 
 		history = []
 		for i, message in enumerate(messages):
@@ -87,9 +134,13 @@ def _build_task_input(self, messages: list[BaseMessage]) -> dict[str, typing.Any
 				history.append(json.dumps(msg))
 			elif message.type == 'human':
 				if len(messages)-1 != i:
+					# Earlier turns keep file parts in history content.
 					history.append(json.dumps({"role": "human", "content": message.content}))
 				else:
-					task_input['input'] = message.content
+					# Current turn: text → input, files → input_attachments.
+					task_input['input'] = extract_text_content(message.content)
+					if self.multimodal:
+						task_input['input_attachments'] = extract_file_ids(message.content)
 			elif message.type == 'tool':
 				content = message.content
 				age = len(messages) - 1 - i
@@ -116,14 +167,16 @@ async def _schedule_task(self, task_input: dict[str, typing.Any], prefer_streami
 
 		await log(nc, LogLvl.DEBUG, task_input)
 
+		task_type = MULTIMODAL_CHAT_WITH_TOOLS if self.multimodal else TEXT_CHAT_WITH_TOOLS
+
 		i = 0
 		while i < 20:
 			try:
 				response = await nc.ocs(
 					"POST",
 					"/ocs/v1.php/taskprocessing/schedule",
 					json={
-						"type": "core:text2text:chatwithtools",
+						"type": task_type,
 						"appId": "context_agent",
 						"input": task_input,
 						"preferStreaming": prefer_streaming,
@@ -180,7 +233,6 @@ def _task_output_text(self, task: Task) -> str | None:
 			return None
 		output = task.output.get('output')
 		return output if isinstance(output, str) else None
-
 	def _raw_task_tool_calls(self, task: Task) -> list[dict[str, typing.Any]]:
 		if not isinstance(task.output, dict):
 			return []
@@ -237,11 +289,15 @@ def _task_to_message(self, task: Task) -> AIMessage:
 		if not isinstance(task.output, dict) or "output" not in task.output:
 			raise Exception('"output" key not found in Nextcloud TaskProcessing task result')
 
+		output_text = task.output['output']
+		output_attachments = task.output.get('output_attachments', [])
+		content = build_multimodal_content(output_text, output_attachments, task.id)
+
 		tool_calls, invalid_tool_calls = self._task_tool_calls(task)
 		if len(tool_calls) > 0 or len(invalid_tool_calls) > 0:
-			message = AIMessage(task.output['output'], tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls)
+			message = AIMessage(content, tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls)
 		else:
-			message = AIMessage(task.output['output'])
+			message = AIMessage(content)
 
 		return message
 
@@ -345,7 +401,21 @@ async def _astream(
 				yield ChatGenerationChunk(message=AIMessageChunk(content=final_delta))
 
 		tool_calls, invalid_tool_calls = self._task_tool_calls(task)
-		if len(tool_calls) > 0 or len(invalid_tool_calls) > 0:
+		output_attachments = []
+		if isinstance(task.output, dict):
+			output_attachments = task.output.get('output_attachments') or []
+
+		if output_attachments:
+			# Append file parts after streamed text so LangChain merges them
+			# into the final AIMessage content (attachments are not streamed).
+			yielded_chunk = True
+			content = [{"type": "file", "file_id": int(file_id), "ocp_task_id": task.id} for file_id in output_attachments]
+			yield ChatGenerationChunk(message=AIMessageChunk(
+				content=content,
+				tool_calls=tool_calls,
+				invalid_tool_calls=invalid_tool_calls,
+			))
+		elif len(tool_calls) > 0 or len(invalid_tool_calls) > 0:
 			yielded_chunk = True
 			yield ChatGenerationChunk(message=AIMessageChunk(content='', tool_calls=tool_calls, invalid_tool_calls=invalid_tool_calls))
 

ex_app/lib/provider.py

@@ -1,25 +1,38 @@
 # SPDX-FileCopyrightText: 2024 Nextcloud GmbH and Nextcloud contributors
 # SPDX-License-Identifier: AGPL-3.0-or-later
-from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider, TaskType, ShapeDescriptor, ShapeType
+from nc_py_api.ex_app.providers.task_processing import TaskProcessingProvider, ShapeDescriptor, ShapeType
 
 
+_optional_output_shape = [
+	ShapeDescriptor(
+		name="sources",
+		description="Used tools",
+		shape_type=ShapeType.LIST_OF_TEXTS
+	)
+]
+
+_optional_input_shape = [
+	ShapeDescriptor(
+		name="memories",
+		description="Injected memories",
+		shape_type=ShapeType.LIST_OF_TEXTS
+	)
+]
+
 provider = TaskProcessingProvider(
 	id='context_agent:agent',
 	name='ContextAgent Provider',
 	task_type='core:contextagent:interaction',
 	expected_runtime=60,
-	optional_output_shape= [
-			ShapeDescriptor(
-				name="sources",
-				description="Used tools",
-				shape_type=ShapeType.LIST_OF_TEXTS
-			)
-	],
-	optional_input_shape= [
-			ShapeDescriptor(
-				name="memories",
-				description="Injected memories",
-				shape_type=ShapeType.LIST_OF_TEXTS
-			)
-	]
-)
+	optional_output_shape=_optional_output_shape,
+	optional_input_shape=_optional_input_shape,
+)
+
+multimodal_provider = TaskProcessingProvider(
+	id='context_agent:agent_multimodal',
+	name='ContextAgent Multimodal Provider',
+	task_type='core:contextagent:multimodal-interaction',
+	expected_runtime=60,
+	optional_output_shape=_optional_output_shape,
+	optional_input_shape=_optional_input_shape,
+)