Agent Zero
9d4f701a25
- New plugin app/plugins/builtins/mcp_server/ with 9 MCP tools
- Tools: search_contacts, get_contact, create_contact, list_calendar_entries,
create_calendar_entry, list_emails, send_email, list_files, upload_file
- Routes: GET /api/v1/mcp/tools, POST /api/v1/mcp/tools/{name}/execute, GET /api/v1/mcp/config
- API-token auth via session + RBAC (mcp:read, mcp:write)
- Frontend: mcp.ts API client with React Query hooks
- Frontend: SettingsMcp.tsx settings page with tool listing and execution
- i18n: de.json and en.json updated with MCP entries
- Tests: 7 tests covering tool listing, config, execution, auth, schema validation
147 lines
4.9 KiB
Python
147 lines
4.9 KiB
Python
"""Tests for MCP Server plugin — tool listing, execution, config.
|
|
|
|
Tests Task 5.16 acceptance criteria.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from tests.conftest import ORIGIN_HEADER
|
|
|
|
|
|
# ─── AC1: List MCP tools ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac1_list_mcp_tools(mcp_authed_client):
|
|
"""AC1: GET /api/v1/mcp/tools → 200 + tool list with 9 tools."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] == 9
|
|
assert len(data["tools"]) == 9
|
|
tool_names = [t["name"] for t in data["tools"]]
|
|
assert "search_contacts" in tool_names
|
|
assert "get_contact" in tool_names
|
|
assert "create_contact" in tool_names
|
|
assert "list_calendar_entries" in tool_names
|
|
assert "create_calendar_entry" in tool_names
|
|
assert "list_emails" in tool_names
|
|
assert "send_email" in tool_names
|
|
assert "list_files" in tool_names
|
|
assert "upload_file" in tool_names
|
|
|
|
|
|
# ─── AC2: Get MCP config ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac2_get_mcp_config(mcp_authed_client):
|
|
"""AC2: GET /api/v1/mcp/config → 200 + server config."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/config", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["server_name"] == "LeoCRM"
|
|
assert data["server_version"] == "1.0.0"
|
|
assert data["protocol_version"] == "2024-11-05"
|
|
assert data["auth_method"] == "api-token"
|
|
assert "search_contacts" in data["available_tools"]
|
|
assert len(data["available_tools"]) == 9
|
|
|
|
|
|
# ─── AC3: Execute search_contacts tool ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac3_execute_search_contacts(mcp_authed_client):
|
|
"""AC3: POST /api/v1/mcp/tools/search_contacts/execute → 200 + results."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/search_contacts/execute",
|
|
json={"arguments": {"query": "Admin", "limit": 10}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["tool"] == "search_contacts"
|
|
assert data["success"] is True
|
|
assert "result" in data
|
|
assert "contacts" in data["result"]
|
|
|
|
|
|
# ─── AC4: Execute non-existent tool returns 404 ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac4_execute_nonexistent_tool(mcp_authed_client):
|
|
"""AC4: POST /api/v1/mcp/tools/nonexistent/execute → 404."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/nonexistent_tool/execute",
|
|
json={"arguments": {}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 404
|
|
assert "tool_not_found" in resp.text
|
|
|
|
|
|
# ─── AC5: Tool definitions have correct schema ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac5_tool_definitions_schema(mcp_authed_client):
|
|
"""AC5: GET /api/v1/mcp/tools → tools have proper schema with parameters."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
tools = resp.json()["tools"]
|
|
|
|
# Check search_contacts has query and limit params
|
|
search_tool = next(t for t in tools if t["name"] == "search_contacts")
|
|
param_names = [p["name"] for p in search_tool["parameters"]]
|
|
assert "query" in param_names
|
|
assert "limit" in param_names
|
|
query_param = next(p for p in search_tool["parameters"] if p["name"] == "query")
|
|
assert query_param["required"] is True
|
|
|
|
# Check create_contact has name, email, phone, type params
|
|
create_tool = next(t for t in tools if t["name"] == "create_contact")
|
|
create_params = [p["name"] for p in create_tool["parameters"]]
|
|
assert "name" in create_params
|
|
assert "email" in create_params
|
|
assert "phone" in create_params
|
|
assert "type" in create_params
|
|
|
|
|
|
# ─── AC6: Unauthorized access is rejected ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac6_unauthorized_access(mcp_client_fixture):
|
|
"""AC6: GET /api/v1/mcp/tools without auth → 401."""
|
|
resp = await mcp_client_fixture.get("/api/v1/mcp/tools", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
# ─── AC7: Execute create_contact tool ───
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ac7_execute_create_contact(mcp_authed_client):
|
|
"""AC7: POST /api/v1/mcp/tools/create_contact/execute → 200 + created contact."""
|
|
client, _ = mcp_authed_client
|
|
resp = await client.post(
|
|
"/api/v1/mcp/tools/create_contact/execute",
|
|
json={"arguments": {"name": "MCP Test Contact", "email": "mcp@test.com", "phone": "+49123456789", "type": "person"}},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["tool"] == "create_contact"
|
|
assert data["success"] is True
|
|
assert data["result"]["name"] == "MCP Test Contact"
|
|
assert data["result"]["email"] == "mcp@test.com"
|