diff --git a/src/mcp_server_appwrite/operator.py b/src/mcp_server_appwrite/operator.py index ce89c9a..f69ba95 100644 --- a/src/mcp_server_appwrite/operator.py +++ b/src/mcp_server_appwrite/operator.py @@ -175,6 +175,8 @@ def get_public_tools(self) -> list[types.Tool]: name="appwrite_search_tools", description=( "Search the hidden Appwrite tool catalog by natural language query. " + "Matches include parameter schemas (name, type, required/optional, " + "description) to use with appwrite_call_tool. " "Use this before appwrite_call_tool when using the Appwrite operator surface." ), input_schema={ @@ -444,9 +446,11 @@ def _search_tools(self, arguments: dict[str, Any]) -> list[ToolContent]: if match.entry.description else "" ) + params = _format_params_block(match.entry) lines.append( f"{index}. tool={match.entry.tool_name} service={match.entry.service_name} " - f"class={match.entry.classification} required={required}{missing} score={match.score}{description}" + f"class={match.entry.classification} required={required}{missing} " + f"score={match.score}{description}{params}" ) lines.append("") lines.append( @@ -622,6 +626,56 @@ def _has_schema_property(entry: CatalogEntry, key: str) -> bool: return isinstance(properties, dict) and key in properties +_PARAM_DESCRIPTION_LIMIT = 120 + + +def _json_schema_type_label(schema: dict[str, Any] | Any) -> str: + if not isinstance(schema, dict): + return "string" + + schema_type = schema.get("type") + if schema_type == "array": + items = schema.get("items") + if isinstance(items, dict): + item_type = items.get("type") + if isinstance(item_type, str) and item_type: + return f"array[{item_type}]" + return "array" + + if isinstance(schema_type, str) and schema_type: + return schema_type + + return "string" + + +def _format_params_block(entry: CatalogEntry) -> str: + properties = entry.input_schema.get("properties") + if not isinstance(properties, dict) or not properties: + return "" + + required = set(entry.required) + ordered_names = [name for name in entry.required if name in properties] + ordered_names.extend(name for name in properties if name not in required) + + lines = ["\n params:"] + for name in ordered_names: + prop_schema = properties.get(name) + if not isinstance(prop_schema, dict): + prop_schema = {} + type_label = _json_schema_type_label(prop_schema) + requirement = "required" if name in required else "optional" + description = str(prop_schema.get("description") or "").strip() + if description: + lines.append( + f" - {name} ({type_label}, {requirement}): " + f"{description[:_PARAM_DESCRIPTION_LIMIT]}" + ) + else: + lines.append(f" - {name} ({type_label}, {requirement})") + + return "\n".join(lines) + + def _compute_score( entry: CatalogEntry, query_tokens: list[str], diff --git a/tests/unit/test_operator.py b/tests/unit/test_operator.py index af3de8d..9def5b2 100644 --- a/tests/unit/test_operator.py +++ b/tests/unit/test_operator.py @@ -8,14 +8,18 @@ def make_tool( - name: str, description: str, required: list[str] | None = None + name: str, + description: str, + required: list[str] | None = None, + properties: dict | None = None, ) -> types.Tool: return types.Tool( name=name, description=description, input_schema={ "type": "object", - "properties": { + "properties": properties + or { "parameter": {"type": "string"}, }, "required": required or [], @@ -61,7 +65,26 @@ def make_runtime(self, executor): }, "tables_db_create": { "definition": make_tool( - "tables_db_create", "Create a database.", ["database_id"] + "tables_db_create", + "Create a database.", + ["database_id", "name"], + properties={ + "database_id": { + "type": "string", + "description": ( + "Unique Id. Choose a custom ID or generate a " + "random ID with `ID.unique()`." + ), + }, + "name": { + "type": "string", + "description": "Database name. Max length: 128 chars.", + }, + "enabled": { + "type": "boolean", + "description": "Is the database enabled?", + }, + }, ), "function": object(), "parameter_types": {}, @@ -71,6 +94,28 @@ def make_runtime(self, executor): "function": object(), "parameter_types": {}, }, + "users_list": { + "definition": make_tool( + "users_list", + "Get a list of all the project's users.", + properties={ + "queries": { + "type": "array", + "items": {"type": "string"}, + "description": ( + "Array of query strings generated using the " + "Query class provided by the SDK." + ), + }, + "search": { + "type": "string", + "description": "Search term to filter your list results.", + }, + }, + ), + "function": object(), + "parameter_types": {}, + }, "functions_create": { "definition": make_tool( "functions_create", @@ -84,7 +129,35 @@ def make_runtime(self, executor): "definition": make_tool( "tables_db_create_string_column", "Create a string column in a table.", - ["database_id", "table_id", "key", "size"], + ["database_id", "table_id", "key", "size", "required"], + properties={ + "database_id": { + "type": "string", + "description": "Database ID.", + }, + "table_id": { + "type": "string", + "description": "Table ID.", + }, + "key": { + "type": "string", + "description": "Column Key.", + }, + "size": { + "type": "number", + "description": ( + "Column size for text columns, in number of characters." + ), + }, + "required": { + "type": "boolean", + "description": "Is column required?", + }, + "default": { + "type": "string", + "description": "Default value for column when not provided.", + }, + }, ), "function": object(), "parameter_types": {}, @@ -211,6 +284,46 @@ def test_search_tools_surfaces_required_create_tool_without_argument_hints(self) self.assertEqual(len(result), 1) self.assertIn("tables_db_create_string_column", result[0].text) + def test_search_tools_includes_parameter_schemas(self): + runtime = self.make_runtime(lambda name, arguments, *_: []) + + users_result = runtime.execute_public_tool( + "appwrite_search_tools", + {"query": "list users", "service_hints": "users"}, + ) + self.assertIn("users_list", users_result[0].text) + self.assertIn("params:", users_result[0].text) + self.assertIn( + "queries (array[string], optional): Array of query strings generated", + users_result[0].text, + ) + + create_result = runtime.execute_public_tool( + "appwrite_search_tools", + {"query": "create database", "service_hints": "tables_db"}, + ) + self.assertIn("tables_db_create", create_result[0].text) + self.assertIn( + "database_id (string, required): Unique Id. Choose a custom ID", + create_result[0].text, + ) + self.assertIn("name (string, required): Database name.", create_result[0].text) + self.assertIn("enabled (boolean, optional)", create_result[0].text) + + column_result = runtime.execute_public_tool( + "appwrite_search_tools", + {"query": "create string column"}, + ) + self.assertIn("tables_db_create_string_column", column_result[0].text) + self.assertIn( + "size (number, required): Column size for text columns", + column_result[0].text, + ) + # Required params are listed before optional ones. + size_pos = column_result[0].text.index("size (number, required)") + default_pos = column_result[0].text.index("default (string, optional)") + self.assertLess(size_pos, default_pos) + def test_search_tools_scores_get_queries_against_get_tools(self): runtime = self.make_runtime(lambda name, arguments, *_: [])