IAC Memory MCP Server

Instructions

Implementation Reference

• src/iac_memory_mcp_server/tools/ansible.py:417-458 (handler)

  MCP tool handler that validates arguments, calls the database update function, logs the operation, and returns success/error messages.

  async def handle_update_module_version(
      db: Any, arguments: Dict[str, Any], operation_id: str
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      """Handle update_module_version tool."""
      try:
          logger.info(
              "Updating module version",
              extra={
                  "module_id": arguments["module_id"],
                  "new_schema": "...",  # Truncated for log
                  "operation_id": operation_id,
              },
          )
  
          # Update module version
          result = update_module_version(
              db,
              arguments["module_id"],
              arguments["new_schema"],
              arguments.get("new_version"),
              arguments.get("new_doc_url"),
          )
  
          return [TextContent(
              type="text",
              text=f"Updated module version: {result}"
          )]
  
      except Exception as e:
          error_msg = f"Failed to update module version: {str(e)}"
          logger.error(error_msg, extra={"operation_id": operation_id})
          raise McpError(
              types.ErrorData(
                  code=types.INTERNAL_ERROR,
                  message=error_msg,
                  data={
                      "tool": "update_module_version",
                      "operation_id": operation_id,
                  },
              )
          )

• src/iac_memory_mcp_server/db/tools.py:277-287 (schema)

  JSON schema defining the input parameters for the update_module_version tool, including required fields like module_id and new_schema.

  "update_module_version": {
      "type": "object",
      "description": "Update an existing Ansible module's schema and related information",
      "required": ["module_id", "new_schema"],
      "properties": {
          "module_id": {"type": "string", "description": "Module ID"},
          "new_schema": {"type": "string", "description": "New schema"},
          "new_version": {"type": "string", "description": "New version"},
          "new_doc_url": {"type": "string", "description": "New documentation URL"},
      },
  },

• src/iac_memory_mcp_server/tools/ansible.py:559-570 (registration)

  Dictionary mapping tool names to their handler functions, registering update_module_version with its handler.

  ansible_tool_handlers = {
      "get_ansible_collection_info": handle_get_ansible_collection_info,
      "list_ansible_collections": handle_list_ansible_collections,
      "get_collection_version_history": handle_get_collection_version_history,
      "get_ansible_module_info": handle_get_ansible_module_info,
      "list_collection_modules": handle_list_collection_modules,
      "get_module_version_compatibility": handle_get_module_version_compatibility,
      "add_ansible_collection": handle_add_ansible_collection,
      "add_ansible_module": handle_add_ansible_module,
      "update_collection_version": handle_update_collection_version,
      "update_module_version": handle_update_module_version,
  }

• src/iac_memory_mcp_server/db/ansible.py:372-410 (helper)

  Database helper function that performs the SQL UPDATE on the ansible_modules table to update schema, version, doc_url, and timestamp.

  def update_module_version(
      db: DatabaseManager,
      module_id: str,
      new_schema: str,
      new_version: Optional[str] = None,
      new_doc_url: Optional[str] = None,
  ) -> bool:
      """Update an Ansible module's schema and optional fields."""
      try:
          with db.get_connection() as conn:
              conn.execute("BEGIN IMMEDIATE")  # Start transaction
              try:
                  updates = ["schema = ?"]
                  params = [new_schema]
  
                  if new_version:
                      updates.append("version = ?")
                      params.append(new_version)
                  if new_doc_url:
                      updates.append("doc_url = ?")
                      params.append(new_doc_url)
  
                  updates.append("updated_at = CURRENT_TIMESTAMP")
                  params.append(module_id)
  
                  cursor = conn.execute(
                      f"""UPDATE ansible_modules
                      SET {', '.join(updates)}
                      WHERE id = ?""",
                      tuple(params),
                  )
                  conn.commit()
                  return cursor.rowcount > 0
              except Exception:
                  conn.rollback()
                  raise
      except sqlite3.Error as e:
          raise DatabaseError(f"Failed to update module version: {str(e)}")