feat(talk): add MCP integration for Nextcloud Talk (spreed)

Adds 6 MCP tools so an LLM can read a user's Talk conversations and
post messages on their behalf, addressing the "read my chats and reply"
use case from issue #720:

  - talk_list_conversations
  - talk_get_conversation
  - talk_get_messages
  - talk_list_participants
  - talk_send_message    (auto-attaches a referenceId for retry dedup)
  - talk_mark_as_read

Edit/delete messages, reactions, threads, and call/session ops are
intentionally out of scope for this first PR.

The TalkClient also exposes create_conversation/delete_conversation
for the integration test fixture; these are not registered as MCP
tools. A post-installation hook enables spreed in the docker dev env
so the integration suite has a real Talk backend to talk to.

Closes #720

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-29 23:19:57 +02:00
co-authored by Claude Opus 4.7
parent a9c5759869
commit 69814f30e3
12 changed files with 1278 additions and 3 deletions
+143
View File
@@ -0,0 +1,143 @@
"""Integration tests for the Nextcloud Talk (spreed) MCP tools."""
import json
import logging
import pytest
from mcp import ClientSession
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
EXPECTED_TALK_TOOLS = {
"talk_list_conversations",
"talk_get_conversation",
"talk_get_messages",
"talk_list_participants",
"talk_send_message",
"talk_mark_as_read",
}
async def test_talk_mcp_connectivity(nc_mcp_client: ClientSession):
"""All six Talk tools should be registered with the MCP server."""
tools = await nc_mcp_client.list_tools()
tool_names = {tool.name for tool in tools.tools}
missing = EXPECTED_TALK_TOOLS - tool_names
assert not missing, f"Missing Talk tools: {missing}"
async def test_talk_send_and_read_workflow(
nc_mcp_client: ClientSession,
nc_client: NextcloudClient,
temporary_conversation: dict,
):
"""End-to-end: post a message via MCP, read it back, mark read."""
token = temporary_conversation["token"]
# 1. Send a message via MCP
send_result = await nc_mcp_client.call_tool(
"talk_send_message",
{"token": token, "message": "Hello from MCP integration test"},
)
assert send_result.isError is False, (
f"talk_send_message failed: {send_result.content}"
)
send_payload = json.loads(send_result.content[0].text)
assert send_payload["success"] is True
posted = send_payload["message"]
assert posted["message"] == "Hello from MCP integration test"
assert posted["token"] == token
posted_id = posted["id"]
logger.info(f"Posted message id={posted_id} into token={token}")
# 2. Cross-check via direct client
direct_messages, _ = await nc_client.talk.get_messages(token, limit=10)
direct_ids = [m.id for m in direct_messages]
assert posted_id in direct_ids, "Posted message not visible via direct client"
# 3. Read messages via MCP
get_result = await nc_mcp_client.call_tool(
"talk_get_messages", {"token": token, "limit": 10}
)
assert get_result.isError is False, (
f"talk_get_messages failed: {get_result.content}"
)
get_payload = json.loads(get_result.content[0].text)
assert get_payload["conversation_token"] == token
listed_ids = [m["id"] for m in get_payload["results"]]
assert posted_id in listed_ids, "Posted message not in MCP get_messages results"
# 4. Mark conversation as read up to that message
mark_result = await nc_mcp_client.call_tool(
"talk_mark_as_read",
{"token": token, "last_read_message": posted_id},
)
assert mark_result.isError is False, (
f"talk_mark_as_read failed: {mark_result.content}"
)
mark_payload = json.loads(mark_result.content[0].text)
assert mark_payload["success"] is True
assert mark_payload["conversation_token"] == token
assert mark_payload["last_read_message"] == posted_id
async def test_talk_list_conversations_includes_temp_room(
nc_mcp_client: ClientSession, temporary_conversation: dict
):
"""Newly created conversation should appear in talk_list_conversations."""
token = temporary_conversation["token"]
list_result = await nc_mcp_client.call_tool("talk_list_conversations", {})
assert list_result.isError is False, (
f"talk_list_conversations failed: {list_result.content}"
)
payload = json.loads(list_result.content[0].text)
tokens = [r["token"] for r in payload["results"]]
assert token in tokens, "Temporary conversation not found in list"
async def test_talk_get_conversation(
nc_mcp_client: ClientSession, temporary_conversation: dict
):
"""talk_get_conversation returns the same room we created."""
token = temporary_conversation["token"]
name = temporary_conversation["name"]
result = await nc_mcp_client.call_tool("talk_get_conversation", {"token": token})
assert result.isError is False, f"talk_get_conversation failed: {result.content}"
payload = json.loads(result.content[0].text)
conversation = payload["conversation"]
assert conversation["token"] == token
assert conversation["name"] == name
async def test_talk_list_participants(
nc_mcp_client: ClientSession, temporary_conversation: dict
):
"""talk_list_participants returns the room creator as a participant."""
token = temporary_conversation["token"]
result = await nc_mcp_client.call_tool("talk_list_participants", {"token": token})
assert result.isError is False, f"talk_list_participants failed: {result.content}"
payload = json.loads(result.content[0].text)
assert payload["conversation_token"] == token
actor_ids = [p["actorId"] for p in payload["results"]]
# The user that created the room is always a participant.
assert len(actor_ids) >= 1
async def test_talk_send_message_validation_empty_text(
nc_mcp_client: ClientSession, temporary_conversation: dict
):
"""Empty message text is rejected client-side."""
token = temporary_conversation["token"]
result = await nc_mcp_client.call_tool(
"talk_send_message", {"token": token, "message": ""}
)
assert result.isError is True, "Expected validation error for empty message"