Appwrite MCP Server

Instructions

Implementation Reference

• src/mcp_server_appwrite/server.py:103-123 (handler)

  Universal handler for all tools. For 'tables_db_create_string_column', it retrieves the bound TablesDB.create_string_column method from the registry and invokes it with user arguments, handling results and errors.

  @server.call_tool()
  async def handle_call_tool(
      name: str, arguments: dict | None
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      
      try:
          tool_info = tools_manager.get_tool(name)
          if not tool_info:
              raise McpError(f"Tool {name} not found")
          
          bound_method = tool_info["function"]
          result = bound_method(**(arguments or {}))
          if hasattr(result, 'to_dict'):
              result_dict = result.to_dict()
              return [types.TextContent(type="text", text=str(result_dict))]
          return [types.TextContent(type="text", text=str(result))]
      except AppwriteException as e:
          return [types.TextContent(type="text", text=f"Appwrite Error: {str(e)}")]
      except Exception as e:
          return [types.TextContent(type="text", text=f"Error: {str(e)}")]

• src/mcp_server_appwrite/service.py:73-127 (schema)

  Dynamically generates tool definitions for all public methods on TablesDB(client), including schema from type hints and docstrings. Constructs tool name as 'tables_db_create_string_column' from service_name='tables_db' and method_name='create_string_column'.

  def list_tools(self) -> Dict[str, Dict]:
      """Lists all available tools for this service"""
      tools = {}
  
      for name, func in inspect.getmembers(self.service, predicate=inspect.ismethod):
          if name.startswith('_'): # Skip private methods
              continue
  
          original_func = func.__func__
          
          # Skip if not from the service's module
          if original_func.__module__ != self.service.__class__.__module__:
              continue
  
          # Get the overridden name if it exists
          tool_name = self._method_name_overrides.get(name, f"{self.service_name}_{name}")
  
          docstring = parse(original_func.__doc__)
          signature = inspect.signature(original_func)
          type_hints = get_type_hints(original_func)
  
          properties = {}
          required = []
  
          for param_name, param in signature.parameters.items():
              if param_name == 'self':
                  continue
  
              param_type = type_hints.get(param_name, str)
              properties[param_name] = self.python_type_to_json_schema(param_type)
              properties[param_name]["description"] = f"Parameter '{param_name}'"
              
              for doc_param in docstring.params:
                  if doc_param.arg_name == param_name:
                      properties[param_name]["description"] = doc_param.description
  
              if param.default is param.empty:
                  required.append(param_name)
  
          tool_definition = Tool(
              name=tool_name,
              description=f"{docstring.short_description or "No description available"}",
              inputSchema={
                  "type": "object",
                  "properties": properties,
                  "required": required
              }
          )
          
          tools[tool_name] = {
              "definition": tool_definition,
              "function": func
          }
          
      return tools

• src/mcp_server_appwrite/server.py:62-95 (registration)

  Registers the TablesDB service (with service_name='tables_db') which triggers introspection and registration of its methods as tools, including 'tables_db_create_string_column'.

  def register_services(args):
      # If --all is specified, enable all services
      if args.all:
          args.tables_db = args.users = args.teams = args.storage = True
          args.functions = args.messaging = args.locale = args.avatars = True
          args.sites = True
  
      # Register services based on CLI arguments
      if args.tables_db:
          tools_manager.register_service(Service(TablesDB(client), "tables_db"))
      if args.users:
          tools_manager.register_service(Service(Users(client), "users"))
      if args.teams:
          tools_manager.register_service(Service(Teams(client), "teams"))
      if args.storage:
          tools_manager.register_service(Service(Storage(client), "storage"))
      if args.functions:
          tools_manager.register_service(Service(Functions(client), "functions"))
      if args.messaging:
          tools_manager.register_service(Service(Messaging(client), "messaging"))
      if args.locale:
          tools_manager.register_service(Service(Locale(client), "locale"))
      if args.avatars:
          tools_manager.register_service(Service(Avatars(client), "avatars"))
      if args.sites:
          tools_manager.register_service(Service(Sites(client), "sites"))
      if args.databases:
          tools_manager.register_service(Service(Databases(client), "databases"))
  
      # If no services were specified, enable tables_db by default
      if not any([args.databases, args.tables_db, args.users, args.teams, args.storage,
                  args.functions, args.messaging, args.locale, args.avatars, args.sites]):
          tools_manager.register_service(Service(TablesDB(client), "tables_db"))

• src/mcp_server_appwrite/tool_manager.py:10-14 (registration)

  ToolManager.register_service appends the service and updates the global tools_registry with the tools from TablesDB service, storing definition and function for 'tables_db_create_string_column'.

  def register_service(self, service: Service):
      """Register a new service and its tools"""
      self.services.append(service)
      self.tools_registry.update(service.list_tools())

• src/mcp_server_appwrite/service.py:23-72 (helper)

  Helper function used to convert Python type hints of TablesDB.create_string_column parameters to JSON Schema for the tool's inputSchema.

  def python_type_to_json_schema(self, py_type: Any) -> dict:
      """Converts Python type hints to JSON Schema types."""
      type_mapping = {
          str: "string",
          int: "integer",
          float: "number",
          bool: "boolean",
          list: "array",
          dict: "object"
      }
      
      # Handle basic types
      if py_type in type_mapping:
          return {"type": type_mapping[py_type]}
      
      # Handle Optional types (Union[type, None])
      if hasattr(py_type, "__origin__") and py_type.__origin__ is Union:
          args = getattr(py_type, "__args__", ())
          if len(args) == 2 and args[1] is type(None):
              schema = self.python_type_to_json_schema(args[0])
              return schema
      
      # Handle List, Dict, and other generic types
      if hasattr(py_type, "__origin__"):
          origin = py_type.__origin__
          args = getattr(py_type, "__args__", ())
          
          # Handle List[T]
          if origin is list or origin is List:
              if args:
                  item_schema = self.python_type_to_json_schema(args[0])
                  return {
                      "type": "array",
                      "items": item_schema
                  }
              return {"type": "array"}
          
          # Handle Dict[K, V]
          if origin is dict or origin is Dict:
              if len(args) >= 2:
                  value_schema = self.python_type_to_json_schema(args[1])
                  return {
                      "type": "object",
                      "additionalProperties": value_schema
                  }
              return {"type": "object"}
      
      # Default to string for unknown types
      return {"type": "string"}