Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 26 additions & 1 deletion src/mcp_server_appwrite/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -831,6 +831,26 @@ def _prepare_arguments(tool_info: dict, arguments: dict[str, Any]) -> dict[str,
return prepared_arguments


def _selects_attributes(arguments: dict[str, Any]) -> bool:
"""Whether the call narrows its response with a ``select`` query."""
queries = arguments.get("queries")
if not isinstance(queries, list):
return False
for query in queries:
if isinstance(query, str):
try:
query = json.loads(query)
except ValueError:
continue
if isinstance(query, dict) and query.get("method") == "select":
return True
return False


def _untyped_response(response: Any, model: Any = None) -> Any:
return response


def _raise_bounded_response_error(response: httpx.Response) -> None:
"""Translate an upstream streaming error into the SDK's public exception."""
body = bytearray()
Expand Down Expand Up @@ -977,7 +997,12 @@ def execute_registered_tool(
hosted = client is None
if client is None:
client = resolve_client(target_project, organization_id)
bound_method = getattr(service_cls(client), method_name)
service = service_cls(client)
if _selects_attributes(prepared_arguments):
# A select query returns only the chosen attributes, which the SDK's
# full response models reject, so the payload is passed through untyped.
setattr(service, "_parse_response", _untyped_response)
bound_method = getattr(service, method_name)
bounded_binary = (
hosted and inspect.signature(bound_method).return_annotation is bytes
)
Expand Down
31 changes: 31 additions & 0 deletions tests/unit/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -1381,6 +1381,37 @@ def get_file_download(self) -> bytes:
client.call.assert_called_once()
self.assertIsInstance(result[0], types.EmbeddedResource)

def test_execute_registered_tool_returns_selected_attributes_untyped(self):
# A select query returns partial documents that the SDK's full models
# reject (Sentry MCP-12/MCP-13).
manager = register_services(object(), profile=OAUTH_PROFILE)
partial = {
"total": 1,
"deployments": [{"$id": "deployment", "type": "cli", "status": "ready"}],
}
select = json.dumps({"method": "select", "values": ["type", "status"]})
client = build_introspection_client()
client.call = Mock(return_value=partial)

result = execute_registered_tool(
manager,
"functions_list_deployments",
{"function_id": "function", "queries": [select]},
client=client,
)

self.assertIsInstance(result[0], types.TextContent)
self.assertEqual(json.loads(result[0].text), partial)

with patch.object(server_module.error_monitoring, "capture_appwrite_exception"):
with self.assertRaises(RuntimeError):
execute_registered_tool(
manager,
"functions_list_deployments",
{"function_id": "function"},
client=client,
)

def test_execute_registered_tool_captures_publishable_appwrite_error(self):
tool = types.Tool(
name="users_list",
Expand Down
Loading